跨境互联网 跨境互联网
首页
  • AI 工具

    • 绘图提示词工具 (opens new window)
    • ChatGPT 指令 (opens new window)
  • ChatGPT

    • ChatGP T介绍
    • ChatGPT API 中文开发手册
    • ChatGPT 中文调教指南
    • ChatGPT 开源项目
  • Midjourney

    • Midjourney 文档
  • Stable Diffusion

    • Stable Diffusion 文档
  • 其他

    • AIGC 热门文章
    • 账号合租 (opens new window)
    • 有趣的网站
  • Vue

    • Vue3前置
  • JAVA基础

    • Stream
    • Git
    • Maven
    • 常用第三方类库
    • 性能调优工具
    • UML系统建模
    • 领域驱动设计
    • 敏捷开发
    • Java 测试
    • 代码规范及工具
    • Groovy 编程
  • 并发编程&多线程

    • 并发编程
    • 高性能队列 Disruptor
    • 多线程并发在电商系统下的应用
  • 其他

    • 面试题
  • 消息中间中间件

    • Kafka
    • RabbitMQ
    • RocketMQ
  • 任务调度

    • Quartz
    • XXL-Job
    • Elastic-Job
  • 源码解析

    • Mybatis 高级使用
    • Mybatis 源码剖析
    • Mybatis-Plus
    • Spring Data JPA
    • Spring 高级使用
    • Spring 源码剖析
    • SpringBoot 高级使用
    • SpringBoot 源码剖析
    • Jdk 解析
    • Tomcat 架构设计&源码剖析
    • Tomcat Web应用服务器
    • Zookeeper 高级
    • Netty
  • 微服务框架

    • 分布式原理
    • 分布式集群架构场景化解决方案
    • Dubbo 高级使用
    • Dubbo 核心源码剖析
    • Spring Cloud Gateway
    • Nacos 实战应用
    • Sentinel 实战应用
    • Seata 分布式事务
  • 数据结构和算法的深入应用
  • 存储

    • 图和Neo4j
    • MongoDB
    • TiDB
    • MySQL 优化
    • MySQL 平滑扩容实战
    • MySQL 海量数据存储与优化
    • Elasticsearch
  • 缓存

    • Redis
    • Aerospike
    • Guava Cache
    • Tair
  • 文件存储

    • 阿里云 OSS 云存储
    • FastDF 文件存储
  • 基础

    • Linux 使用
    • Nginx 使用与配置
    • OpenResty 使用
    • LVS+Keepalived 高可用部署
    • Jekins
  • 容器技术

    • Docker
    • K8S
    • K8S
  • 01.全链路(APM)
  • 02.电商终极搜索解决方案
  • 03.电商亿级数据库设计
  • 04.大屏实时计算
  • 05.分库分表的深入实战
  • 06.多维系统下单点登录
  • 07.多服务之间分布式事务
  • 08.业务幂等性技术架构体系
  • 09.高并发下的12306优化
  • 10.每秒100W请求的秒杀架构体系
  • 11.集中化日志管理平台的应用
  • 12.数据中台配置中心
  • 13.每天千万级订单的生成背后痛点及技术突破
  • 14.红包雨的架构设计及源码实现
  • 人工智能

    • Python 笔记
    • Python 工具库
    • 人工智能(AI) 笔记
    • 人工智能(AI) 项目笔记
  • 大数据

    • Flink流处理框架
  • 加密区

    • 机器学习(ML) (opens new window)
    • 深度学习(DL) (opens new window)
    • 自然语言处理(NLP) (opens new window)
AI 导航 (opens new window)

Revin

首页
  • AI 工具

    • 绘图提示词工具 (opens new window)
    • ChatGPT 指令 (opens new window)
  • ChatGPT

    • ChatGP T介绍
    • ChatGPT API 中文开发手册
    • ChatGPT 中文调教指南
    • ChatGPT 开源项目
  • Midjourney

    • Midjourney 文档
  • Stable Diffusion

    • Stable Diffusion 文档
  • 其他

    • AIGC 热门文章
    • 账号合租 (opens new window)
    • 有趣的网站
  • Vue

    • Vue3前置
  • JAVA基础

    • Stream
    • Git
    • Maven
    • 常用第三方类库
    • 性能调优工具
    • UML系统建模
    • 领域驱动设计
    • 敏捷开发
    • Java 测试
    • 代码规范及工具
    • Groovy 编程
  • 并发编程&多线程

    • 并发编程
    • 高性能队列 Disruptor
    • 多线程并发在电商系统下的应用
  • 其他

    • 面试题
  • 消息中间中间件

    • Kafka
    • RabbitMQ
    • RocketMQ
  • 任务调度

    • Quartz
    • XXL-Job
    • Elastic-Job
  • 源码解析

    • Mybatis 高级使用
    • Mybatis 源码剖析
    • Mybatis-Plus
    • Spring Data JPA
    • Spring 高级使用
    • Spring 源码剖析
    • SpringBoot 高级使用
    • SpringBoot 源码剖析
    • Jdk 解析
    • Tomcat 架构设计&源码剖析
    • Tomcat Web应用服务器
    • Zookeeper 高级
    • Netty
  • 微服务框架

    • 分布式原理
    • 分布式集群架构场景化解决方案
    • Dubbo 高级使用
    • Dubbo 核心源码剖析
    • Spring Cloud Gateway
    • Nacos 实战应用
    • Sentinel 实战应用
    • Seata 分布式事务
  • 数据结构和算法的深入应用
  • 存储

    • 图和Neo4j
    • MongoDB
    • TiDB
    • MySQL 优化
    • MySQL 平滑扩容实战
    • MySQL 海量数据存储与优化
    • Elasticsearch
  • 缓存

    • Redis
    • Aerospike
    • Guava Cache
    • Tair
  • 文件存储

    • 阿里云 OSS 云存储
    • FastDF 文件存储
  • 基础

    • Linux 使用
    • Nginx 使用与配置
    • OpenResty 使用
    • LVS+Keepalived 高可用部署
    • Jekins
  • 容器技术

    • Docker
    • K8S
    • K8S
  • 01.全链路(APM)
  • 02.电商终极搜索解决方案
  • 03.电商亿级数据库设计
  • 04.大屏实时计算
  • 05.分库分表的深入实战
  • 06.多维系统下单点登录
  • 07.多服务之间分布式事务
  • 08.业务幂等性技术架构体系
  • 09.高并发下的12306优化
  • 10.每秒100W请求的秒杀架构体系
  • 11.集中化日志管理平台的应用
  • 12.数据中台配置中心
  • 13.每天千万级订单的生成背后痛点及技术突破
  • 14.红包雨的架构设计及源码实现
  • 人工智能

    • Python 笔记
    • Python 工具库
    • 人工智能(AI) 笔记
    • 人工智能(AI) 项目笔记
  • 大数据

    • Flink流处理框架
  • 加密区

    • 机器学习(ML) (opens new window)
    • 深度学习(DL) (opens new window)
    • 自然语言处理(NLP) (opens new window)
AI 导航 (opens new window)
  • 任务调度

  • 消息队列

    • 消息中间件(MQ)介绍
    • Kafka

      • Kafka基础使用
        • 1、应用场景
          • 1.1 kafka场景
          • 1.2 kafka特性
          • 1.3 消息对比
          • 1.4 大厂应用
        • 2、基础组件
          • 2.1 角色
          • 2.2 逻辑组件
          • 2.3 副本集合
          • 2.4 消息标记
        • 3、架构探索
          • 3.1 发展历程
          • 3.1.1 版本命名
          • 3.1.2 演进历史
          • 3.2 集群搭建(助学)
          • 3.3 组件探秘
          • 3.2.1 主题创建
          • 3.2.2 查看主题
          • 3.2.3 主题详情
          • 3.2.4 消息收发
          • 3.2.5 分组消费
          • 3.2.6 指定分区
          • 3.2.7 偏移量
          • 3.4 zk探秘
          • 3.4.1 broker信息
          • 3.4.2 主题与分区
          • 3.4.3 消费者与偏移量
          • 3.4.4 controller
          • 3.5 km
          • 3.5.1 启动
          • 3.5.2 使用
        • 4、深入应用
          • 4.1 springboot-kafka
          • 4.2 消息发送
          • 4.2.1 发送类型
          • 4.2.2 序列化
          • 4.2.3 分区策略
          • 4.3 消息消费
          • 4.3.1 消息组别
          • 4.3.2 位移提交
        • 5、高级特性
          • 5.1 扩展性
          • 5.1.1 broker扩容
          • 5.1.2 分区扩容
          • 5.2 高可用
          • 5.2.1 节点选举
          • 5.2.2 原理剖析
        • 6、底层架构
          • 6.1 存储架构
          • 6.1.1 分段存储
          • 6.1.2 日志索引
          • 6.1.3 日志删除
          • 6.1.4 存储验证
          • 6.2 零拷贝
          • 6.2.1 传统文件读写
          • 6.2.2 DMA
          • 6.2.3 java实现
          • 6.3 分区一致性
          • 6.3.1 水位值
          • 6.3.2 同步原理
          • 6.3.3 Leader Epoch
        • 7、业务实战
          • 7.1 顺序性场景
          • 7.1.1 场景概述
          • 7.1.2 顺序级别
          • 7.1.3 实现方案
          • 7.1.4 代码验证
          • 7.2 海量同步场景
          • 7.2.1 常规架构
          • 7.2.2 解耦合
          • 7.2.3 部署实现
          • 7.3 kafka监控
          • 7.3.1 eagle简介
          • 7.3.2 部署
          • 7.3.3 使用说明
      • Kafka深入

    • RabbitMQ

    • RocketMQ

  • Zookeeper

  • java组件
  • 消息队列
  • Kafka
Revin
2023-06-16
目录

Kafka基础使用

hm

# 1、应用场景

# 1.1 kafka场景

Kafka最初是由LinkedIn公司采用Scala语言开发,基于ZooKeeper,现在已经捐献给了Apache基金会。目前Kafka已经定位为一个分布式流式处理平台,它以 高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

Apache Kafka能够支撑海量数据的数据传递。在离线和实时的消息处理业务系统中,Kafka都有广泛的应用。

(1)日志收集:收集各种服务的log,通过kafka以统一接口服务的方式开放 给各种consumer,例如Hadoop、Hbase、Solr等;

(2)消息系统:解耦和生产者和消费者、缓存消息等;

(3)用户活动跟踪:Kafka经常被用来记录web用户或者app用户的各种活动,如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到kafka的topic中,然后订阅者通过订阅这些topic来做实时 的监控分析,或者装载到Hadoop、数据仓库中做离线分析和挖掘;

(4)运营指标:Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据,生产各种操作的集中反馈,比如报警和报告;

(5)流式处理:比如spark streaming和storm;

# 1.2 kafka特性

kafka以高吞吐量著称,主要有以下特性:

(1)高吞吐量、低延迟:kafka每秒可以处理几十万条消息,它的延迟最低只有几毫秒;

(2)可扩展性:kafka集群支持热扩展;

(3)持久性、可靠性:消息被持久化到本地磁盘,并且支持数据备份防止数据丢失;

(4)容错性:允许集群中节点失败(若副本数量为n,则允许n-1个节点失败);

(5)高并发:支持数千个客户端同时读写;

# 1.3 消息对比

  • 如果普通的业务消息解耦,消息传输,rabbitMq是首选,它足够简单,管理方便,性能够用。
  • 如果在上述,日志、消息收集、访问记录等高吞吐,实时性场景下,推荐kafka,它基于分布式,扩容便捷
  • 如果很重的业务,要做到极高的可靠性,考虑rocketMq,但是它太重。需要你有足够的了解

# 1.4 大厂应用

  • 京东通过kafka搭建数据平台,用于用户购买、浏览等行为的分析。成功抗住6.18的流量洪峰
  • 阿里借鉴kafka的理念,推出自己的rocketmq。在设计上参考了kafka的架构体系

# 2、基础组件

# 2.1 角色

image-20210104101417979

  • broker:节点,就是你看到的机器
  • provider:生产者,发消息的
  • consumer:消费者,读消息的
  • zookeeper:信息中心,记录kafka的各种信息的地方
  • controller:其中的一个broker,作为leader身份来负责管理整个集群。如果挂掉,借助zk重新选主

# 2.2 逻辑组件

image-20210104112057008

  • topic:主题,一个消息的通道,收发总得知道消息往哪投
  • partition:分区,每个主题可以有多个分区分担数据的传递,多条路并行,吞吐量大
  • Replicas:副本,每个分区可以设置多个副本,副本之间数据一致。相当于备份,有备胎更可靠
  • leader & follower:主从,上面的这些副本里有1个身份为leader,其他的为follower。leader处理partition的所有读写请求

# 2.3 副本集合

  • AR(assigned replica):所有副本的统称,AR=ISR+OSR
  • ISR(In-sync Replica):同步中的副本,可以参与leader选主。一旦落后太多(数量滞后和时间滞后两个维度)会被踢到OSR。
  • OSR(Out-Sync Relipcas):踢出同步的副本,一直追赶leader,追上后会进入ISR

# 2.4 消息标记

image-20210104111840162

image-20210104140818843

  • offset:偏移量,消息消费到哪一条了?每个消费者都有自己的偏移量
  • HW:(high watermark):副本的高水印值,客户端最多能消费到的位置,HW值为8,代表offset为[0,8]的9条消息都可以被消费到,它们是对消费者可见的,而[9,12]这4条消息由于未提交,对消费者是不可见的。
  • LEO:(log end offset):日志末端位移,代表日志文件中下一条待写入消息的offset,这个offset上实际是没有消息的。不管是leader副本还是follower副本,都有这个值。

那么这三者有什么关系呢?

比如在副本数等于3的情况下,消息发送到Leader A之后会更新LEO的值,Follower B和Follower C也会实时拉取Leader A中的消息来更新自己,HW就表示A、B、C三者同时达到的日志位移,也就是A、B、C三者中LEO最小的那个值。由于B、C拉取A消息之间延时问题,所以HW一般会小于LEO,即LEO>=HW。

具体的同步原理,下面章节会详细讲到

# 3、架构探索

# 3.1 发展历程

http://kafka.apache.org/downloads

image-20210104142336618

# 3.1.1 版本命名

Kafka在1.0.0版本前的命名规则是4位,比如0.8.2.1,0.8是大版本号,2是小版本号,1表示打过1个补丁

现在的版本号命名规则是3位,格式是“大版本号”+“小版本号”+“修订补丁数”,比如2.5.0,前面的2代表的是大版本号,中间的5代表的是小版本号,0表示没有打过补丁

我们所看到的下载包,前面是scala编译器的版本,后面才是真正的kafka版本。

# 3.1.2 演进历史

0.7版本 只提供了最基础的消息队列功能。

0.8版本 引入了副本机制,至此Kafka成为了一个真正意义上完备的分布式高可靠消息队列解决方案。

0.9版本 增加权限和认证,使用Java重写了新的consumer API,Kafka Connect功能;不建议使用consumer API;

0.10版本 引入Kafka Streams功能,正式升级成分布式流处理平台;建议版本0.10.2.2;建议使用新版consumer API

0.11版本 producer API幂等,事务API,消息格式重构;建议版本0.11.0.3;谨慎对待消息格式变化

1.0和2.0版本 Kafka Streams改进;建议版本2.0;

# 3.2 集群搭建(助学)

1)原生启动

kafka启动需要zookeeper,第一步启动zk:

docker run --name zookeeper-1 -d -p 2181 zookeeper:3.4.13
1

原生安装:下载后解压启动即可 http://kafka.apache.org/downloads

bin/kafka-server-start.sh config/server.properties
#server.properties配置说明
#表示broker的编号,如果集群中有多个broker,则每个broker的编号需要设置的不同
broker.id=0 
#brokder对外提供的服务入口地址,默认9092
listeners=PLAINTEXT://:9092 
#设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log 
#Kafka所需Zookeeper集群地址,这里是关键!加入同一个zk的kafka为同一集群
zookeeper.connect=zookeeper:2181 
1
2
3
4
5
6
7
8
9
10

2)推荐docker-compose 一键启动

#参考资料中的kafka.yml
#注意hostname问题,ip地址:192.168.10.30,换成你自己服务器的
#docker-compose -f kafka.yml up -d 启动
version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13


    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
        environment:
            KAFKA_BROKER_ID: 1 
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10903 
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper           
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
        environment:
            KAFKA_BROKER_ID: 2 
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10904 
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# 3.3 组件探秘

命令行工具是管理kafka集群最直接的工具。官方自带,不需要额外安装。

# 3.2.1 主题创建

image-20211210142749419

#进入容器
docker exec -it kafka-1 sh
#进入bin目录
cd /opt/kafka/bin
#创建
kafka-topics.sh --zookeeper zookeeper:2181 --create --topic test --partitions 2 --replication-factor 1
1
2
3
4
5
6

# 3.2.2 查看主题

kafka-topics.sh --zookeeper zookeeper:2181 --list
1

# 3.2.3 主题详情

kafka-topics.sh --zookeeper zookeeper:2181 --describe --topic test


#分析输出:
Topic:test  PartitionCount:2    ReplicationFactor:1 Configs:
    Topic: test Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: test Partition: 1    Leader: 1   Replicas: 1 Isr: 1
1
2
3
4
5
6
7

# 3.2.4 消息收发

#使用docker连接任意集群中的一个容器
docker exec -it kafka-1 sh


#进入kafka的容器内目录
cd /opt/kafka/bin


#客户端监听
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test


#另起一个终端,验证发送
./kafka-console-producer.sh --broker-list localhost:9092 --topic test


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 3.2.5 分组消费

#启动两个consumer时,如果不指定group信息,消息被广播
#指定相同的group,让多个消费者分工消费(画图:group原理)


./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group aaa


#结果:在发送方,连续发送 1-4 ,4条消息,同一group下的两台consumer交替消费,并发执行
1
2
3
4
5
6
7
8

注意!!!

这是在消费者和分区数相等(都是2)的情况下。 如果同一group下的 ( 消费者数量 > 分区数量 ) 那么就会有消费者闲置。

验证方式:

可以再多启动几个消费者试一试,会发现,超出2个的时候,有的始终不会消费到消息。 停掉可以消费到的,那么闲置的会被激活,进入工作状态

# 3.2.6 指定分区

#指定分区通过参数 --partition,注意!需要去掉上面的group
#指定分区的意义在于,保障消息传输的顺序性(画图:kafka顺序性原理)
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 1


#结果:发送1-4条消息,交替出现。说明消息被均分到各个分区中投递




#默认的发送是没有指定key的
#要指定分区发送,就需要定义key。那么相同的key被路由到同一个分区
./kafka-console-producer.sh --broker-list kafka-1:9092 --topic test --property parse.key=true


#携带key再发送,注意key和value之间用tab分割
>1  1111
>1  2222
>2  3333
>2  4444


#查看consumer的接收情况
#结果:相同的key被同一个consumer消费掉


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 3.2.7 偏移量

#偏移量决定了消息从哪开始消费,支持:开头,还是末尾


## earliest: 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
## latest: 当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
## none: topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常


## 注意点!!!有提交偏移量的话,仍然以提交的为主,即便使用earliest,比提交点更早的也不会被提取


#--offset [earliest|latest(默认)] , 或者 --from-beginning
#新起一个终端,指定offset位置
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset earliest


./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --from-beginning


#结果:之前发送的消息,从头又消费了一遍!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 3.4 zk探秘

前面说过,zk存储了kafka集群的相关信息,本节来探索内部的秘密。

kafka的信息记录在zk中,进入zk容器,查看相关节点和信息

docker exec -it kafka-zookeeper-1 sh


>./bin/zkCli.sh


>ls /


#结果:得到以下配置信息
1
2
3
4
5
6
7
8
9
10

image-20210104183056471

# 3.4.1 broker信息

[zk: localhost:2181(CONNECTED) 0] ls /brokers
[ids, topics, seqid]
[zk: localhost:2181(CONNECTED) 1] ls /brokers/ids
[1, 2]


#机器broker信息
[zk: localhost:2181(CONNECTED) 4] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://192.168.10.30:10903"],"jmx_port":-1,"host":"192.168.10.30","timestamp":"1609825245500","port":10903,"version":4}
cZxid = 0x27
ctime = Tue Jan 05 05:40:45 GMT 2021
mZxid = 0x27
mtime = Tue Jan 05 05:40:45 GMT 2021
pZxid = 0x27
cversion = 0
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x105a2db626b0000
dataLength = 196
numChildren = 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 3.4.2 主题与分区

#分区节点路径
[zk: localhost:2181(CONNECTED) 5] ls /brokers/topics
[test, __consumer_offsets]
[zk: localhost:2181(CONNECTED) 6] ls /brokers/topics/test
[partitions]
[zk: localhost:2181(CONNECTED) 7] ls /brokers/topics/test/partitions
[0, 1]
[zk: localhost:2181(CONNECTED) 8] ls /brokers/topics/test/partitions/0
[state]


#分区信息,leader所在的机器id,isr列表等
[zk: localhost:2181(CONNECTED) 18] get /brokers/topics/test/partitions/0/state
{"controller_epoch":1,"leader":1,"version":1,"leader_epoch":0,"isr":[1]}
cZxid = 0xb0
ctime = Tue Jan 05 05:56:06 GMT 2021
mZxid = 0xb0
mtime = Tue Jan 05 05:56:06 GMT 2021
pZxid = 0xb0
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 72
numChildren = 0


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 3.4.3 消费者与偏移量

[zk: localhost:2181(CONNECTED) 15] ls /consumers
[]
#空的???
#那么,消费者以及它的偏移记在哪里呢???


1
2
3
4
5
6

kafka 消费者记录 group 的消费 偏移量 有两种方式 :

1)kafka 自维护 (新)

2)zookpeer 维护 (旧) ,已经逐渐被废弃

查看方式:

上面的消费用的是控制台工具,这个工具使用--bootstrap-server,不经过zk,也就不会记录到/consumers下。

其消费者的offset会更新到一个kafka自带的topic【__consumer_offsets】下面

#先起一个消费端,指定group
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group aaa


#使用控制台工具查看消费者及偏移量情况
./kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --list
KMOffsetCache-44acff134cad
aaa


#查看偏移量详情
./kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group aaa
1
2
3
4
5
6
7
8
9
10
11
12

当前与LEO保持一致,说明消息都完整的被消费过

image-20210105174238632

停掉consumer后,往provider中再发几条记录,offset开始滞后:

image-20210105174424989

重新启动consumer,消费到最新的消息,同时再返回看偏移量,消息得到同步:

image-20210105174658407

# 3.4.4 controller

#当前集群中的主控节点是谁
[zk: localhost:2181(CONNECTED) 17] get /controller
{"version":1,"brokerid":1,"timestamp":"1609825245694"}
cZxid = 0x2a
ctime = Tue Jan 05 05:40:45 GMT 2021
mZxid = 0x2a
mtime = Tue Jan 05 05:40:45 GMT 2021
pZxid = 0x2a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x105a2db626b0000
dataLength = 54
numChildren = 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 3.5 km

# 3.5.1 启动

kafka-manager是目前最受欢迎的kafka集群管理工具,最早由雅虎开源。提供可视化kafka集群操作

官网:https://github.com/yahoo/kafka-manager/releases

注意它的版本,docker社区的镜像版本滞后于kafka,我们自己来打镜像。

#Dockerfile
FROM daocloud.io/library/java:openjdk-8u40-jdk
ADD kafka-manager-2.0.0.2/ /opt/km2002/
CMD ["/opt/km2002/bin/kafka-manager","-Dconfig.file=/opt/km2002/conf/application.conf"]


#打包,注意将kafka-manager-2.0.0.2放到同一目录
docker build -t km:2002 .


## 还可以直接拉取
docker pull liggdocker/km:2002
## 修改镜像标签为km:2002
docker tag imageId km:2002
#启动:在上面的yml里,services节点下加一段
#参考资料:km.yml
#执行: docker-compose -f km.yml up -d
        km:
        image: liggdocker/km:2002
        ports:
            - 10906:9000
        depends_on:
            - zookeeper
         
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

完整的km.yml内容

version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13


    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
        environment:
            KAFKA_BROKER_ID: 1
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10903
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
        environment:
            KAFKA_BROKER_ID: 2
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10904
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper
    km:
        image: liggdocker/km:2002
        ports:
            - 10906:9000
        depends_on:
            - zookeeper


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# 3.5.2 使用

使用km可以方便的查看以下信息:

  • cluster:创建集群,填写zk地址,选中jmx,consumer信息等选项
  • brokers:列表,机器信息
  • topic:主题信息,主题内的分区信息。创建新的主题,增加分区
  • cosumers: 消费者信息,偏移量等

# 4、深入应用

# 4.1 springboot-kafka

1)配置文件

  kafka:
    bootstrap-servers: 192.168.10.30:10903,192.168.10.30:10904
    producer: # producer 生产者
      retries: 0 # 重试次数
      acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      batch-size: 16384 # 一次最多发送数据量
      buffer-memory: 33554432 # 生产端缓冲区大小
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer


    consumer: # consumer消费者
      group-id: javagroup # 默认的消费组ID
      enable-auto-commit: true # 是否自动提交offset
      auto-commit-interval: 100 # 提交offset延时(接收到消息后多久提交offset)
      auto-offset-reset: latest  #earliest,latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2)启动信息

image-20210106143550399

# 4.2 消息发送

# 4.2.1 发送类型

KafkaTemplate调用send时默认采用异步发送,如果需要同步获取发送结果,调用get方法

详细代码参考:AsyncProducer.java

消费者使用:KafkaConsumer.java

1)同步发送

        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("test", JSON.toJSONString(message));
                //注意,可以设置等待时间,超出后,不再等候结果
        SendResult<String, Object> result = future.get(3,TimeUnit.SECONDS);
        logger.info("send result:{}",result.getProducerRecord().value());
1
2
3
4

通过swagger发送,控制台可以正常打印send result

swagger访问地址:http://localhost:8080/doc.html

2)阻断

在服务器上,将kafka暂停服务

docker-compose -f km.yml pause kafka-1 kafka-2
1

在swagger发送消息

调同步发送:请求被阻断,一直等待,超时后返回错误

image-20210106155900539

而调异步发送的(默认发送接口),请求立刻返回。

image-20210106160001017

那么,异步发送的消息怎么确认发送情况呢???往下看!

3)注册监听

代码参考: KafkaListener.java (释放注解)

可以给kafkaTemplate设置Listener来监听消息发送情况,实现内部的对应方法

 kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {});
1

查看控制台,等待一段时间后,异步发送失败的消息会被回调给注册过的listener

com.itheima.demo.config.KafkaListener:error!message={"message":"1","sendTime":1609920296374}
1

启动kafka

docker-compose unpause kafka-1 kafka-2
1

再次发送消息时,同步异步均可以正常收发,并且监听进入success回调

com.itheima.demo.config.KafkaListener$1:ok,message={"message":"1","sendTime":1610089315395}
com.itheima.demo.controller.PartitionConsumer:patition=1,message:[{"message":"1","sendTime":1610089315395}]
1
2

可以看到,在内部类 KafkaListener$1 中,即注册的Listener的消息。

# 4.2.2 序列化

消费者使用:KafkaConsumer.java

1)序列化详解

  • 前面用到的是Kafka自带的字符串序列化器(org.apache.kafka.common.serialization.StringSerializer)
  • 除此之外还有:ByteArray、ByteBuffer、Bytes、Double、Integer、Long 等
  • 这些序列化器都实现了接口 (org.apache.kafka.common.serialization.Serializer)
  • 基本上,可以满足绝大多数场景

2)自定义序列化

自己实现,实现对应的接口即可,有以下方法:

public interface Serializer<T> extends Closeable {
    default void configure(Map<String, ?> configs, boolean isKey) {
    }


    //理论上,只实现这个即可正常运行
    byte[] serialize(String var1, T var2);


    //默认调上面的方法
    default byte[] serialize(String topic, Headers headers, T data) {
        return this.serialize(topic, data);
    }


    default void close() {
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

案例,参考: MySerializer.java

在yaml中配置自己的编码器

value-serializer: com.itheima.demo.config.MySerializer
1

重新发送,发现:消息发送端编码回调一切正常。但是消费端消息内容不对!

com.itheima.demo.controller.KafkaListener$1:ok,message={"message":"1","sendTime":1609923570477}
com.itheima.demo.controller.KafkaConsumer:message:"{\"message\":\"1\",\"sendTime\":1609923570477}"
1
2

怎么办?

3)解码

发送端有编码并且我们自己定义了编码,那么接收端自然要配备对应的解码策略

代码参考:MyDeserializer.java,实现方式与编码器几乎一样!

在yaml中配置自己的解码器

value-deserializer: com.itheima.demo.config.MyDeserializer
1

再次收发,消息正常

com.itheima.demo.controller.AsyncProducer$1:ok,message={"message":"1","sendTime":1609924855896}
com.itheima.demo.controller.KafkaConsumer:message:{"message":"1","sendTime":1609924855896}
1
2

# 4.2.3 分区策略

分区策略决定了消息根据key投放到哪个分区,也是顺序消费保障的基石。

  • 给定了分区号,直接将数据发送到指定的分区里面去
  • 没有给定分区号,给定数据的key值,通过key取上hashCode进行分区
  • 既没有给定分区号,也没有给定key值,直接轮循进行分区
  • 自定义分区,你想怎么做就怎么做

1)验证默认分区规则

发送者代码参考:PartitionProducer.java

消费者代码使用:PartitionConsumer.java

通过swagger访问setKey:

image-20210108134831705

看控制台:

image-20210108134950158

再访问setPartition来设置分区号0来发送

image-20210108135041543

看控制台:

image-20210108135148546

2)自定义分区

你想自己定义规则,根据我的要求,把消息投放到对应的分区去? 可以!

参考代码:MyPartitioner.java , MyPartitionTemplate.java ,

发送使用:MyPartitionProducer.java

使用swagger,发送0开头和非0开头两种key试一试!

image-20210108152145108

备注:

自己定义config参数,比较麻烦,需要打破默认的KafkaTemplate设置

可以将KafkaConfiguration.java中的getTemplate加上@Bean注解来覆盖系统默认bean

这里为了避免混淆,采用@Autowire注入

# 4.3 消息消费

# 4.3.1 消息组别

发送者使用:KafkaProducer.java

1)代码参考:GroupConsumer.java,Listener拷贝3份,分别赋予两组group,验证分组消费:

image-20210106142254297

2)启动

image-20210106143956316

3)通过swagger发送2条消息

image-20210106144226686

  • 同一group下的两个消费者,在group1均分消息
  • group2下只有一个消费者,得到全部消息

4)消费端闲置

注意分区数与消费者数的搭配,如果 ( 消费者数 > 分区数量 ),将会出现消费者闲置,浪费资源!

验证方式:

停掉项目,删掉test主题,重新建一个 ,这次只给它分配一个分区。

重新发送两条消息,试一试

image-20210205171103705

解析:

group2可以消费到1、2两条消息

group1下有两个消费者,但是只分配给了 -1 , -2这个进程被闲置

# 4.3.2 位移提交

1)自动提交

前面的案例中,我们设置了以下两个选项,则kafka会按延时设置自动提交

enable-auto-commit: true # 是否自动提交offset
auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)
1
2

2)手动提交

有些时候,我们需要手动控制偏移量的提交时机,比如确保消息严格消费后再提交,以防止丢失或重复。

下面我们自己定义配置,覆盖上面的参数

代码参考:MyOffsetConfig.java

通过在消费端的Consumer来提交偏移量,有如下几种方式:

代码参考:MyOffsetConsumer.java

同步提交、异步提交:manualCommit() ,同步异步的差别,下面会详细讲到。

指定偏移量提交:offset()

3)重复消费问题

如果手动提交模式被打开,一定不要忘记提交偏移量。否则会造成重复消费!

代码参考和对比:manualCommit() , noCommit()

验证过程:

用km将test主题删除,新建一个test空主题。方便观察消息偏移 注释掉其他Consumer的Component注解,只保留当前MyOffsetConsumer.java 启动项目,使用swagger的KafkaProducer发送连续几条消息 留心控制台,都能消费,没问题:

image-20210113120859072

但是!重启试试:

image-20210113121017561

无论重启多少次,不提交偏移量的消费组,会重复消费一遍!!!

再通过命令行查询偏移量试试:

image-20210113121449839

4)经验与总结

commitSync()方法,即同步提交,会提交最后一个偏移量。在成功提交或碰到无怯恢复的错误之前,commitSync()会一直重试,但是commitAsync()不会。


这就造成一个陷阱:
如果异步提交,针对偶尔出现的提交失败,不进行重试不会有太大问题,因为如果提交失败是因为临时问题导致的,那么后续的提交总会有成功的。只要成功一次,偏移量就会提交上去。


但是!如果这是发生在关闭消费者时的最后一次提交,就要确保能够提交成功,如果还没提交完就停掉了进程。就会造成重复消费!


因此,在消费者关闭前一般会组合使用commitAsync()和commitSync()。
详细代码参考:MyOffsetConsumer.manualOffset()
1
2
3
4
5
6
7
8
9
10
11
12

# 5、高级特性

# 5.1 扩展性

# 5.1.1 broker扩容

1)在yaml中复制kafka-2,拷贝为新的节点,注意以下标注修改的地方!

#修改后的内容参考:cluster.yml


    kafka-3: #改
        container_name: kafka-3 #改
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10905:9092 #改
        environment:
            KAFKA_BROKER_ID: 3 #改
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10905 #改
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

完整的 cluster.yml

version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13


    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
        environment:
            KAFKA_BROKER_ID: 1
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10903
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
        environment:
            KAFKA_BROKER_ID: 2
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10904
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper
    km:
        image: liggdocker/km:2002
        ports:
            - 10906:9000
        depends_on:
            - zookeeper
    kafka-3: #改
        container_name: kafka-3 #改
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10905:9092 #改
        environment:
            KAFKA_BROKER_ID: 3 #改
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10905 #改
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

2)更新docker集群信息

docker-compose -f cluster.yml up -d
#启动消息


kafka_zookeeper_1 is up-to-date
kafka_km_1 is up-to-date
kafka-1 is up-to-date
kafka-2 is up-to-date
Creating kafka-3 ... done
1
2
3
4
5
6
7
8
9

3)进命令行,或打开km查看新的broker信息

image-20210114123502766

# 5.1.2 分区扩容

1)使用km对test主题增加分区到3个,看分区分配机器情况

image-20210114123626581

可以指定新分区数量,及分配到的机器

image-20210114123856288

2)注意问题

新加分区或重新调整分区,已经启动的客户端会动态更新对应的分配信息,不需要重启。

但是!!!

在同步变更消息的过程中有可能会丢失消息!想想为什么?(答案在下面)

(注意!以下场景不保证100%会重现!)

image-20210114130806920

答案:

回顾一下消费偏移量的默认提交配置:latest,因为新分区没有任何offset提交记录

所以会在重新分配分区后从末尾开始消费!

那么分配前的那些消息就不会消费到。而分配后再发送的不会受影响,可以正常消费

分区分配正常后,查看偏移量提交信息,没问题:

image-20210114125922113

km的Consumer页签里也可以查看偏移量信息:

image-20210114161441582

# 5.2 高可用

以上动态扩容操作是怎么实现的呢?集群中必然有一个节点协调了相关操作。

这台协调者,就是controller节点。

controller节点是其中的一台broker,所有broker都有可能成为controller

当前controller宕机后,其他就会参与竞争,选出新的controller,保持集群对外的高可用

# 5.2.1 节点选举

1)查找controller,找到它所在的broker

#查找docker进程,找到zookeeper的容器
[root@iZ8vb3a9qxofwannyywl6zZ ~]# docker ps --format "table{{.ID}}\t{{.Names}}\t{{.Ports}}"
CONTAINER ID        NAMES               PORTS
75318748caab        kafka-3             0.0.0.0:10905->9092/tcp
4807d188a180        kafka_km_1          0.0.0.0:10906->9000/tcp
4453eb0b2a36        kafka-2             0.0.0.0:10904->9092/tcp
d6fd814a0851        kafka-1             0.0.0.0:10903->9092/tcp
8c1fc2cc6e9a        kafka_zookeeper_1   2181/tcp, 2888/tcp, 3888/tcp


#进入容器,连上zk
[root@iZ8vb3a9qxofwannyywl6zZ ~]# docker exec -it kafka_zookeeper_1 sh
/zookeeper-3.4.13 #
/zookeeper-3.4.13 # zkCli.sh
Connecting to localhost:2181


#查询当前controller是哪个节点,发现是2号机器(有可能是其他节点,找到这个brokerid,下面要用!)
[zk: localhost:2181(CONNECTED) 6] get /controller
{"version":1,"brokerid":2,"timestamp":"1610500701187"}


#controller变更的次数
[zk: localhost:2181(CONNECTED) 7] get /controller_epoch
1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

​

2)docker-compose停掉它!

#docker pause 暂停容器的服务,注意是上面找到的那台broker
[root@iZ8vb3a9qxofwannyywl6zZ ~]# docker pause kafka-2
kafka-2


#查看状态,发现(Paused)
[root@iZ8vb3a9qxofwannyywl6zZ ~]# docker ps | grep kafka-2
4453eb0b2a36        wurstmeister/kafka:2.12-2.2.2                    "start-kafka.sh"         2 days ago          Up 2 days (Paused)   0.0.0.0:10904->9092/tcp                                                                      kafka-2


#再次按 1)的步骤进入zk容器,查看当前controller,已经变为3号
[zk: localhost:2181(CONNECTED) 0] get /controller
{"version":1,"brokerid":3,"timestamp":"1610679583216"}


#变更次数加了1
[zk: localhost:2181(CONNECTED) 1] get /controller_epoch
2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 5.2.2 原理剖析

当控制器被关闭或者与Zookeeper系统断开连接时,Zookeeper系统上的/controller临时节点就会被清除。

Kafka集群中的监听器会接收到变更通知,各个代理节点会尝试到Zookeeper系统中创建它。

第一个成功在Zookeeper系统中创建的代理节点,将会成为新的控制器。

每个新选举出来的控制器,会在Zookeeper系统中递增controller_epoch的值。

附:详细流程图

img

# 6、底层架构

# 6.1 存储架构

# 6.1.1 分段存储

开篇讲过,kafka每个主题可以有多个分区,每个分区在它所在的broker上创建一个文件夹

每个分区又分为多个段,每个段两个文件,log文件里顺序存消息,index文件里存消息的索引

段的命名直接以当前段的第一条消息的offset为名

注意是偏移量,不是序号! 第几条消息 = 偏移量 + 1。类似数组长度和下标。

所以offset从0开始(可以开新队列新groupid消费第一条消息打印offset得到验证)

image-20210119150603724

例如:

0.log -> 有8条,offset为 0-7,[0, 8)

8.log -> 有两条,offset为 8-9,[8, 10)

10.log -> 有xx条,offset从10-xx,[10, 10 + xx)

img

# 6.1.2 日志索引

每个log文件配备一个索引文件 *.index

/opt/kafka/bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files test-0/00000000000000000000.index

image-20210115162535648

综合上述,来看一个消息的查找:

  • consumer发起请求要求从offset=6的消息开始消费
  • kafka直接根据文件名大小,发现6号消息在00000.log这个文件里
  • 那文件找到了,它在文件的哪个位置呢?
  • 根据index文件,发现 (6 , 9807),说明消息藏在这里!
  • 从log文件的 9807 位置开始读取。
  • 那读多长呢?简单,读到下一条消息的偏移量停止就可以了

# 6.1.3 日志删除

Kafka作为消息中间件,数据需要按照一定的规则删除,否则数据量太大会把集群存储空间占满。

删除数据方式:

  • 按照时间,超过一段时间后删除过期消息
  • 按照消息大小,消息数量超过一定大小后删除最旧的数据

Kafka删除数据的最小单位:segment,也就是直接干掉文件!一删就是一个log和index文件

# 6.1.4 存储验证

1)数据准备

将broker 2和3 停掉,只保留1

docker pause kafka-2 kafka-3
1

2)删掉test主题,通过km新建一个test主题,加2个分区

新建时,注意下面的选项:

segment.bytes = 1000 ,即:每个log文件到达1000byte时,开始创建新文件

删除策略:

retention.bytes = 2000,即:超出2000byte的旧日志被删除

retention.ms = 60000,即:超出1分钟后的旧日志被删除

以上任意一条满足,就会删除。

3)进入kafka-1这台容器

docker exec -it kafka-1 sh


#查看容器中的文件信息
/ # ls /
bin    dev    etc    home   kafka  lib    lib64  media  mnt    opt    proc   root   run    sbin   srv    sys    tmp    usr    var


/ # cd /kafka/


/kafka # ls
kafka-logs-d0b9c75080d6


/kafka # cd kafka-logs-d0b9c75080d6/
/kafka/kafka-logs-d0b9c75080d6 # ls -l | grep test
drwxr-xr-x    2 root     root          4096 Jan 15 14:35 test-0
drwxr-xr-x    2 root     root          4096 Jan 15 14:35 test-1


#2个分区的日志文件清单,注意当前还没有任何消息写进来
#timeindex:日志的时间信息
#leader-epoch,下面会讲到
/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0:
total 4
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root             0 Jan 15 14:35 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


test-1:
total 4
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root             0 Jan 15 14:35 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

4)往里灌数据。启动项目通过swagger发送消息

注意!边发送边查看上一步的文件列表信息!

image-20210115144735864

#先发送2条,消息开始进来,log文件变大!消息在两个分区之间逐个增加。
/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root           875 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


test-1:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root           875 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


#继续逐条发送,返回再来看文件,大小为1000,到达边界!
/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


test-1:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


#继续发送消息!1号分区的log文件开始分裂
#说明第8条消息已经进入了第二个log
/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 14:35 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root      10485756 Jan 15 14:35 00000000000000000000.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


test-1:
total 20
-rw-r--r--    1 root     root             0 Jan 15 14:46 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root            12 Jan 15 14:46 00000000000000000000.timeindex
-rw-r--r--    1 root     root      10485760 Jan 15 14:46 00000000000000000008.index
-rw-r--r--    1 root     root           125 Jan 15 14:46 00000000000000000008.log   #第二个log文件!
-rw-r--r--    1 root     root            10 Jan 15 14:46 00000000000000000008.snapshot
-rw-r--r--    1 root     root      10485756 Jan 15 14:46 00000000000000000008.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


#持续发送,另一个分区也开始分离
/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0: 
total 20
-rw-r--r--    1 root     root             0 Jan 15 15:55 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root            12 Jan 15 15:55 00000000000000000000.timeindex
-rw-r--r--    1 root     root      10485760 Jan 15 15:55 00000000000000000008.index
-rw-r--r--    1 root     root           625 Jan 15 15:55 00000000000000000008.log
-rw-r--r--    1 root     root            10 Jan 15 15:55 00000000000000000008.snapshot
-rw-r--r--    1 root     root      10485756 Jan 15 15:55 00000000000000000008.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint


test-1:
total 20
-rw-r--r--    1 root     root             0 Jan 15 14:46 00000000000000000000.index
-rw-r--r--    1 root     root          1000 Jan 15 14:46 00000000000000000000.log
-rw-r--r--    1 root     root            12 Jan 15 14:46 00000000000000000000.timeindex
-rw-r--r--    1 root     root      10485760 Jan 15 14:46 00000000000000000008.index
-rw-r--r--    1 root     root           750 Jan 15 15:55 00000000000000000008.log
-rw-r--r--    1 root     root            10 Jan 15 14:46 00000000000000000008.snapshot
-rw-r--r--    1 root     root      10485756 Jan 15 14:46 00000000000000000008.timeindex
-rw-r--r--    1 root     root             8 Jan 15 14:35 leader-epoch-checkpoint




#持续发送消息,分区越来越多。
#过一段时间后再来查看,清理任务将会执行,超出的日志被删除!(默认调度间隔5min)
#log.retention.check.interval.ms 参数指定


/kafka/kafka-logs-d0b9c75080d6 # ls -lR test-*
test-0:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 19:12 00000000000000000119.index
-rw-r--r--    1 root     root             0 Jan 15 19:12 00000000000000000119.log
-rw-r--r--    1 root     root            10 Jan 15 19:12 00000000000000000119.snapshot
-rw-r--r--    1 root     root      10485756 Jan 15 19:12 00000000000000000119.timeindex
-rw-r--r--    1 root     root            10 Jan 15 19:12 leader-epoch-checkpoint


test-1:
total 8
-rw-r--r--    1 root     root      10485760 Jan 15 19:12 00000000000000000119.index
-rw-r--r--    1 root     root             0 Jan 15 19:12 00000000000000000119.log
-rw-r--r--    1 root     root            10 Jan 15 19:12 00000000000000000119.snapshot
-rw-r--r--    1 root     root      10485756 Jan 15 19:12 00000000000000000119.timeindex
-rw-r--r--    1 root     root            10 Jan 15 19:12 leader-epoch-checkpoint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

# 6.2 零拷贝

Kafka 在执行消息的写入和读取这么快,其中的一个原因是零拷贝(Zero-copy)技术

# 6.2.1 传统文件读写

img

传统读写,涉及到 4 次数据的复制。但是这个过程中,数据完全没有变化,我们仅仅是想从磁盘把数据送到网卡。

那有没有办法不绕这一圈呢?让磁盘和网卡之类的外围设备直接访问内存,而不经过cpu?

有! 这就是DMA(Direct Memory Access 直接内存访问)。

# 6.2.2 DMA

DMA其实是由DMA芯片(硬件支持)来控制的。通过DMA控制芯片,可以让网卡等外部设备直接去读取内存,而不是由cpu来回拷贝传输。这就是所谓的零拷贝

目前计算机主流硬件基本都支持DMA,就包括我们的硬盘和网卡。

kafka就是调取操作系统的sendfile,借助DMA来实现零拷贝数据传输的

image-20210115210450506

# 6.2.3 java实现

为加深理解,类比为java中的零拷贝:

  • 在Java中的零拷贝是通过java.nio.channels.FileChannel中的transferTo方法来实现的

  • transferTo方法底层通过native调操作系统的sendfile

  • 操作系统sendfile负责把数据从某个fd(linux file descriptor)传输到另一个fd

    备注:linux下所有的设备都是一个文件描述符fd

代码参考:

File file = new File("0.log");
RandomAccessFile raf = new RandomAccessFile(file, "rw");
//文件通道,来源
FileChannel fileChannel = raf.getChannel();
//网络通道,去处
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("1.1.1.1", 1234));
//对接上,通过transfer直接送过去
fileChannel.transferTo(0, fileChannel.size(), socketChannel);
1
2
3
4
5
6
7
8

# 6.3 分区一致性

# 6.3.1 水位值

1)先回顾两个值:

img

2)再看下几个值的存储位置:

注意!分区是有leader和follower的,最新写的消息会进入leader,follower从leader不停的同步

无论leader还是follower,都有自己的HW和LEO,存储在各自分区所在的磁盘上

leader多一个Remote LEO,它表示针对各个follower的LEO,leader又额外记了一份!

3)为什么这么做呢?

leader会拿这些remote值里最小的来更新自己的hw,具体过程我们详细往下看

# 6.3.2 同步原理

img

我们来看这几个值是如何更新的:

1)leader.LEO

这个很简单,每次producer有新消息发过来,就会增加

2)其他值

另外的4个值初始化都是 0

他们的更新由follower的fetch(同步消息线程)得到的数据来决定!

如果把fetch看做是leader上提供的方法,由follower远程请求调用,那么它的伪代码大概是这个样子:

//java伪代码!
//follower端的操作,不停的请求从leader获取最新数据
class Follower{
  private List<Message> messages;
  private HW hw;
  private LEO leo;
  
  @Schedule("不停的向leader发起同步请求")
  void execute(){
    //向leader发起fetch请求,将自己的leo传过去
    //leader返回leo之后最新的消息,以及leader的hw
    LeaderReturn lr = leader.fetch(this.leo) ;
    
    //存消息
    this.messages.addAll(lr.newMsg);
    //增加follower的leo值
    this.leo = this.leo + lr.newMsg.length;
    //比较自己的leo和leader的hw,取两者小的,作为follower的hw
    this.hw = min(this.leo , lr.leaderHW);
  }
}






//leader返回的报文
class LeaderReturn{
  //新增的消息
  List<Messages> newMsg;
  //leader的hw
  HW leaderHW;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
//leader在接到follower的fetch请求时,做的逻辑
class Leader{
  private List<Message> messages;
  private LEO leo;
  private HW hw;
  //Leader比follower多了个Remote!
  //注意!如果有多个副本,那么RemoteLEO也有多个,每个副本对应一个
  private RemoteLEO remoteLEO;
  
  //接到follower的fetch请求时,leader做的事情
  LeaderReturn fetch(LEO followerLEO){
    //根据follower传过来的leo,来更新leader的remote
    this.remoteLEO = followerLEO ;
    //然后取ISR(所有可用副本)的最小leo作为leader的hw
    this.hw = min(this.leo , this.remoteLEO) ;
    
    //从leader的消息列表里,查找大于follower的leo的所有新消息
    List<Message> newMsg = queryMsg(followerLEO) ;
    
    //将最新的消息(大于follower leo的那些),以及leader的hw返回给follower
    LeaderReturn lr = new LeaderReturn(newMsg , this.hw)
    return lr;
  }
  
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 6.3.3 Leader Epoch

1)产生的背景

0.11版本之前的kafka,完全借助hw作为消息的基准,不管leo。

发生故障后的规则:

  • follower故障再次恢复后,从磁盘读取hw的值并从hw开始剔除后面的消息,并同步leader消息
  • leader故障后,新当选的leader的hw作为新的分区hw,其余节点按照此hw进行剔除数据,并重新同步
  • 上述根据hw进行数据恢复会出现数据丢失和不一致的情况,下面分开来看

假设:

我们有两个副本:leader(A),follower(B)

场景一:丢数据

image-20210127152137598

  • 某个时间点B挂了。当它恢复后,以挂之前的hw为准,设置 leo = hw
  • 这就造成一个问题:现实中,leo 很可能是 大于 hw的。leo被回退了!
  • 如果这时候,恰恰A也挂掉了。kafka会重选leader,B被选中。
  • 过段时间,A恢复后变成follower,从B开始同步数据。
  • 问题来了!上面说了,B的数据是被回退过的,以它为基准会有问题
  • 最终结果:两者的数据都发生丢失,没有地方可以找回!

场景二:数据不一致

image-20210127152230174

  • 这次假设AB全挂了。比较惨
  • B先恢复。但是它的hw有可能挂之前没从A同步过来(原来A是leader)
  • 我们假设,A.hw = 2 , B.hw = 1
  • B恢复后,集群里只有它自己,所以被选为leader,开始接受新消息
  • B.hw上涨,变成2
  • 然后,A恢复,原来A.hw = 2 ,恢复后以B的hw,也就是2为基准开始同步。
  • 问题来了!B当leader后新接到的2号消息是不会同步给A的,A一直保留着它当leader时的旧数据
  • 最终结果:数据不一致了!

2)改进思路

0.11之后,kafka改进了hw做主的规则,这就是leader epoch

leader epoch给leader节点带了一个版本号,类似于乐观锁的设计。

它的思想是,一旦发生机器故障,重启之后,不再机械的将leo退回hw

而是借助epoch的版本信息,去请求当前leader,让它去算一算leo应该是什么

3)实现原理

对比上面丢数据的问题:

image-20210117160544546

  • A为(leo=2 , hw=2),B为(leo=2 , hw=1)
  • B重启,但是B不再着急将leo打回hw,而是发起一个Epoch请求给当前leader,也就是A
  • A收到LE=0后,发现和自己的LE一样,说明B在挂掉前后,leader没变,都是A自己
  • 那么A就将自己的leo值返回给B,也就是数字2
  • B收到2后和自己的leo比对取较小值,发现也是2,那么不再退回到hw的1
  • 没有回退,也就是信息1的位置没有被覆盖,最大程度的保护了数据
  • 如果和上面一样的场景,A挂掉,B被选为leader

image-20210117161903544

  • 那么A再次启动时后,从B开始同步数据

  • 因为B之前没有回退,1号信息得到了保留

  • 同时,B的LE(epoch号码)开始增加,从0变成1,offset记录为B当leader时的位置,也就是2

  • A传过来的epoch为0,B是1,不相等。那么取大于0的所有epoch里最小的

    (现实中可能发生了多次重新选主,有多条epoch)

  • 其实就是LE=1的那条。现实中可能有多条。并找到它对应的offset(也就是2)给A返回去

  • 最终A得到了B同步过来的数据

再来看一致性问题的解决:

image-20210117163116528

  • 还是上面的场景,AB同时挂掉,但是hw还没同步,那么A.hw=2 , B.hw=1

  • B先启动被选成了leader,新leader选举后,epoch加了一条记录(参考下图,LE=1,这时候offset=1)

  • 表示B从1开始往后继续写数据,新来了条信息,内容为m3,写到1号位

  • A启动前,集群只有B自己,消息被确认,hw上涨到2,变成下面的样子

    image-20210117163347006

  • A开始恢复,启动后向B发送epoch请求,将自己的LE=0告诉leader,也就是B

  • B发现自己的LE不同,同样去大于0的LE里最小的那条,也就是1 , 对应的offset也是1,返回给A

  • A从1开始同步数据,将自己本地的数据截断、覆盖,hw上升到2

  • 那么最新的写入的m3从B给同步到了A,并覆盖了A上之前的旧数据m2

  • 结果:数据保持了一致

附:epochRequest的详细流程图

processing OffsetForLeaderEpochRequest

# 7、业务实战

# 7.1 顺序性场景

# 7.1.1 场景概述

假设我们要传输一批订单到另一个系统,那么订单对应状态的演变是有顺序性要求的。

已下单 → 已支付 → 已确认

不允许错乱!

# 7.1.2 顺序级别

1)全局有序:

串行化。每条经过kafka的消息必须严格保障有序性。

这就要求kafka单通道,每个groupid下单消费者

极大的影响性能,现实业务下几乎没必要

2)局部有序:

业务局部有序。同一条订单有序即可,不同订单可以并行处理。不同订单的顺序前后无所谓

充分利用kafka多分区的并发性,只需要想办法让需要顺序的一批数据进同一分区即可。

# 7.1.3 实现方案

1)发送端:

指定key发送,key=order.id即可,案例回顾:4.2.3,PartitionProducer

2)发送中:

给队列配置多分区保障并发性。

3)读取端:

单消费者:显然不合理

吞吐量显然上不去,kafka开多个分区还有何意义?

所以开多个消费者指定分区消费,理想状况下,每个分区配一个。

但是,这个吞吐量依然有限,那如何处理呢?

方案:多线程

在每个消费者上再开多线程,是个解决办法。但是,要警惕顺序性被打破!

参考下图:thread处理后,会将data变成 2-1-3

watermark1233212312324

改进:接收后分发二级内存队列

消费者取到消息后不做处理,根据key二次分发到多个阻塞队列。

再开启多个线程,每个队列分配一个线程处理。提升吞吐量

sfsf8u98wwd

# 7.1.4 代码验证

1)新建一个sort队列,2个分区

2)启动order项目

源码参考:

SortedProducer(顺序性发送端)

SortedConsumer(顺序性消费端 - 阻塞队列实现,方便大家理解设计思路)

SortedConsumer2(顺序性消费端 - 线程池实现,现实中推荐这种方式!)

3)通过swagger请求

image-20210120135200524

先按不同的id发送,查看控制台日志,id被正确分发到对应的队列

image-20210120141510550

同一个key分配到同一个queue,顺序性得到保障

image-20210120141712904

# 7.2 海量同步场景

假设大数据部门需要大屏来展示用户的打车订单情况,需要把订单数据送入druid

这里不涉及顺序,只要下单就传输,但是对实时性和并发量要求较高

# 7.2.1 常规架构

在下单完成mysql后,通过程序代码打印,直接进入kafka

或者logback和kafka集成,通过log输送

优点:

更符合常规的思维。将数据送给想要的部门

缺点:

耦合度高,将kafka发送消息嵌入了订单下单的主业务,形成代码入侵。

下单不关心,也不应该关注送入kafka的情况,一旦kafka不可用,程序受影响

# 7.2.2 解耦合

借助canal,监听订单表的数据变化,不再影响主业务。

img

# 7.2.3 部署实现

1)mysql部署

注意,需要打开binlog,8.0 默认处于开启状态

#启动mysql8
docker run --name mysql8 -v /opt/kafka/data/mysql8:/var/lib/mysql -p 3306:3306 -e TZ=Asia/Shanghai -e MYSQL_ROOT_PASSWORD=123456 -d daocloud.io/mysql:8.0
1
2

连上mysql,执行以下sql,添加canal用户

CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
1
2
3
4

创建订单表

CREATE TABLE `orders` (
  `id` int unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
);
1
2
3
4
5

2)canal部署

#canal.properties
#附带资料里有,放到服务器 /opt/kafka/data/canal/ 目录下
#修改servers为你的kafka的机器地址
canal.serverMode = kafka
kafka.bootstrap.servers = 192.168.10.30:10903,192.168.10.30:10904
#docker-compose.yml
#附带资料里有canal.yml,随便找个目录,重命名为docker-compose.yml
#修改mysql的链接信息的链接信息
#然后在当前目录下执行 docker-compose up -d
version: '2'
services:
    canal:
        image: canal/canal-server
        container_name: canal
        restart: always
        ports:
            - "10908:11111"
        environment:
                #mysql的链接信息
            canal.instance.master.address: 192.168.10.30:3306
            canal.instance.dbUsername: canal
            canal.instance.dbPassword: canal
            #投放到kafka的哪个主题?要提前准备好!
            canal.mq.topic: canal
        volumes:
            - "/opt/kafka/data/canal/canal.properties:/home/admin/canal-server/conf/canal.properties"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

3)数据通道验证

进入kafka容器,用上面3.2.4里的命令行方式监听canal队列

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic canal
1

在mysql上创建orders表,增删数据试一下

mysql> insert into orders (name) values ('张三');
Query OK, 1 row affected (0.03 sec)
1
2

在kafka控制台,可以看到同步的消息

{"data":[{"id":"1","name":"张三"}],"database":"canal","es":1611657853000,"id":5,"isDdl":false,"mysqlType":{"id":"int unsigned","name":"varchar(255)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12},"table":"orders","ts":1611657853802,"type":"INSERT"}
1

数据通道已打通,还缺少的是druid作为消费端来接收消息

4)druid部署

#druid.yml
#在附带资料里有
#随便找个目录,执行
docker-compose -f druid.yml up -d
1
2
3
4

5)验证

配置druid的数据源,从kafka读取数据,验证数据可以正确进入druid。

image-20210126185051954

注:

关于druid的详细使用,在大数据篇章里会详细讲解。

# 7.3 kafka监控

# 7.3.1 eagle简介

Kafka Eagle监控系统是一款用来监控Kafka集群的工具,支持管理多个Kafka集群、管理Kafka主题(包含查看、删除、创建等)、消费者组合消费者实例监控、消息阻塞告警、Kafka集群健康状态查看等。

img

# 7.3.2 部署

推荐docker-compose启动

将配备的资料中 eagle.yml , 拷贝到服务器任意目录

修改对应的ip地址为你服务器的地址

#注意ip地址:192.168.10.30,全部换成你自己服务器的


version: '3'
services:
    zookeeper:
        image: zookeeper:3.4.13


    kafka-1:
        container_name: kafka-1
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10903:9092
            - 10913:10913
        environment:
            KAFKA_BROKER_ID: 1 
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            #docker部署必须设置外部可访问ip和端口,否则注册进zk的地址将不可达造成外部无法连接
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10903 
            KAFKA_JMX_OPTS: "-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=192.168.10.30 -Dcom.sun.management.jmxremote.rmi.port=10913"
            JMX_PORT: 10913
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper           
    kafka-2:
        container_name: kafka-2
        image: wurstmeister/kafka:2.12-2.2.2
        ports:
            - 10904:9092
            - 10914:10914
        environment:
            KAFKA_BROKER_ID: 2 
            HOST_IP: 192.168.10.30
            KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
            KAFKA_ADVERTISED_HOST_NAME: 192.168.10.30
            KAFKA_ADVERTISED_PORT: 10904 
            KAFKA_JMX_OPTS: "-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=192.168.10.30 -Dcom.sun.management.jmxremote.rmi.port=10914"
            JMX_PORT: 10914
        volumes:
            - /etc/localtime:/etc/localtime
        depends_on:
            - zookeeper 
    eagle:
        image: gui66497/kafka_eagle
        container_name: ke
        restart: always
        depends_on:
            - kafka-1
            - kafka-2
        ports:
            - "10907:8048"
        environment:
            ZKSERVER: "zookeeper:2181"


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

执行 docker-compose -f eagle.yml up -d

# 7.3.3 使用说明

访问 : http://192.168.10.30:10907/ke/

默认用户名密码: admin/ 123456

如果要删除topic等操作,需要管理token: keadmin

image-20210127105615559

与km到底选哪个呢?根据自己习惯,个人认为:

  • 界面美观程度和监控曲线优于km,有登录权限控制
  • 功能操作上不如km简单直白,但是km需要配置一定的连接信息
上次更新: 2025/04/03, 11:07:08
消息中间件(MQ)介绍
Kafka架构与实战

← 消息中间件(MQ)介绍 Kafka架构与实战→

最近更新
01
tailwindcss
03-26
02
PaddleSpeech
02-18
03
whisper
02-18
更多文章>
Theme by Vdoing | Copyright © 2019-2025 跨境互联网 | 豫ICP备14016603号-5 | 豫公网安备41090002410995号
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式