Kafka 架构和机制、常用脚本工具

Kafka 架构和机制

Zookeeper: 存储元数据
Broker互为主备，Topic按分区存储，Replicas分布在不同节点

节点角色

• Controller：负责Partition管理和Replicas管理，也执行重分配Partition之类的管理任务。若故障，则从其他Broker中重新选举。Broker节点状态管理：新增/下线节点、元数据更新。Topic分区管理：新增/删除topic、分区扩容/迁移/leader切换。-
• Leader：负责某Topic某Partition的读写请求。若故障，则从Follower中重新选举。
• Follower：同步Leader数据。
• Coordinator：负责管理Consumer group，维护消费队列进度。

zookeeper节点

• /controller：中央控制器brokerid等信息
• /brokers/ids/{brokerid, 1-n}：临时znode，broker唯一编号
• /topics/{topic-name}/partitions/{partitionid, 0-n}/state 持久znode，存储分区leader的brokerId等
• /isr_change_notifiction：ISR变更事件通知

topic创建机制

1. 执行创建topic动作
2. 在/config/topics上注册topic的配置目录
3. 在/brokers/topic上注册topic的元数据目录
4. 监听到新topic写入
5. Controller从zk读取topic分区信息
6. 发送分区副本信息到broker
7. broker读取topic配置并在本地创建分区副本

topic删除机制

1. /admin/delete_topics注册监听
2. 开始删除topic线程
3. 阻塞并等待删除事件
4. 恢复删除线程
5. 添加/admin/delete_topics/{topic_name}
6. 删除所有分区、删除zk相关目录、清理contoller相关cache

leader选举流程

创建topic时选举

1. 写入分区信息 /brokers/topics/{topic_name}
2. Controller监听到/brokers/topics变化
3. 读取分区副本列表，首个副本选举成为leader
4. 发送LeaderAndIsr请求到broker节点

Leader失效时选举

1. Controller监听/brokers/ids变化
2. 对leader在该broker上的所有partition重新选举
3. 获取分区isr，选举首个可用节点作为分区新leader
4. 发送LeaderAndIsr请求到broker节点

分区扩容流程

1. 修改/brokers/topics/{topic_name}
2. Controller监听到/brokers/topics变化
3. 读取分区信息，发送分区创建请求到broker
4. broker读取topic配置并在本地创建分区副本

重分区和重新分配：可能需要进行重分区和重新分配。这涉及到数据的重新分布和重新平衡，可能会导致一段时间内的性能下降和延迟增加。

副本迁移流程

先扩容新副本再下线旧副本

1. 写入/admin/reassign_partition
2. Controller监听到副本迁移
3. Controller注册/brokers/topics/{topic_name}/partitions/{partitionid}/state
4. 发送请求到新副本所在broker
5. broker读取topic配置并创建本地副本
6. 新副本从原有leader同步数据
7. 新副本加入ISR，修改zk上分区状态
8. 选举新leader，停止老副本，使用新副本
   分区副本迁移不会中断该分区的生产、消费请求

若Topic为单副本，扩容期间无法对该Topic生产消息或消费消息，会造成业务中断。

kafka utils

可–help

kafka-topics.sh

• create
• delete
• list
• describe
• topic
• partition
• replication-factor
• config

示例：

kafka-topics.sh --bootstrap-server xx:9091 --list
kafka-topics.sh --bootstrap-server xx:9091 --topic toppicName --describe
kafka-topics.sh --bootstrap-server xx:9091 --topic toppicName --create --parition 3 --replication-factor 2
kafka-topics.sh --bootstrap-server xx:9091 --topic toppicName --delete

kafka-console-produer.sh kafka-console-consumer.sh

• kafka-console-consumer.sh –from-beginning

示例：

kafka-topics.sh --bootstrap-server xx:9091 --topic toppicName --group myGroup --from-beginning

kafka-consumer-groups.sh

• bootstrap-server
• command-config
• describe
• group
• list

DMS监控

集群/节点/队列级别监控
消费监控

Donate