Kafka集群管理&生产消费流程

  • Kafka 集群如何进行管理？管理包含元数据存储，topic创建、broker、生产者、消费者的变更等：
  1. 元数据的存储和流转
  2. broker加入、宕机，Kafka 怎么知道，并且周知给集群其他broker的
  3. 消费组中消费者的新增、减少，如何做 rebalance，分派消费者和partition的
  4. 控制器、协调器、leader消费者等 的作用和工作原理
  • Kafka 的生产者、消费者的处理流程介绍

Kafka 集群管理

Controller 是如何选举出来的

  1. broker 把自己的broker id 注册到zk /controller 节点，第一个注册进来的就是 Controller

    其他未抢到的，就是watch 这个节点，以备 controller 挂了之后，进行选举新的

  2. 选举出来的controller 要设置各种 ZK 监听器，来管理集群

普通broker 如何注册的

  1. broker 把自己的broker id 注册到zk /brokers/ids/id号 节点，value包含ip+port等信息

各个broker 如何感知新注册进来的broker

  1. broker 把自己的broker id 注册到zk /brokers/ids/id号 节点
  2. controller会监听 /brokers/ids/ 目录变化，得知新的节点加入进来，然后controller 再把这些元数据信息发给其他broker，每个broker 都要维护所有的broker 信息

broker实例挂了，分区选举

当某个broker 宕机了，此时分布在这个broker 的leader副本就要做重新选举，又由于生产和消费端都要操作leader副本，因此也要感知新的leader副本。

  1. broker 宕机，对应的zk失去节点id
  2. 控制器Controller 感知到该 broker 下线，进行该broker上 leader 分区的选举
  3. 从各个leader 分区对应的 ISR 副本中获取第一个副本，选举成为新的 leader副本
  4. 控制器Controller 再把元数据下发给所有其他broker

创建一个topic的流程 集群如何感知的

  1. 计算分配方案（各partition的leader和副本放哪些个broker上）
  2. 把分配方案写到ZK /brokers/topics/topic名字 目录的value
  3. controller会监听这个目录变更，自然知道这个topic的分配方案，然后controller再把这个元数据分发给其他broker

为什么要有协调器 Coordinator，什么作用，如何选择出来的？

为什么要有 coordinator？

一个消费者组，里面多个消费者，消费某个topic，那每个消费者怎么配合的，消费哪些partition，这个分派的制定方案（leader consumer 负责）需要周知给各个消费者，需要有人来协调的。另外集群中感知消费者存活状态，心跳线程的交互对象。这都需要协调器的角色。

不同于控制器controller的粒度是整个kafka集群，协调器的粒度是消费者组的，当然也是一个broker实例负责的。

coordinator 如何选择出来的？

  1. 每个消费者组有一个group_id，取这个组id的哈希值
  2. 位移topic: __consumer_offsets，默认 50 个partition
  3. 哈希值对50取模，得到的数字代表是哪一个位移分片。看这个分片的leader副本在哪个broker上，这个broker即coordinator

为什么 coordinator 要和位移topic的某个leader partition在同一个broker

这是为了方便，因为消费者在消费时，提交位移要写到这个位移topic对应的partition，而提交位移也正是和 coordinator交互。

leader consumer 是什么？以及作用

消费者组里的每个消费者都会去 coordinator 上注册，第一个注册的即 leader consumer。

leader consumer 负责制定分区的消费方案。包括在rebalance重平衡时，进行重新分派。

制定好消费方案后，会再告诉 coordinator，coordinator再下发给各个消费者。

疑问

为什么需要 leader consumer 制定消费方案？ coordinator 知道所有消费者，由 coordinator制定不就可以了么？

生产、消费流程

生产者发消息流程

默认一个生产者对应一个topic，那消息发送到这个topic对应的哪个partition，要和哪个broker通信？即要先获取到这个topic的元数据信息。

  1. 获取该topic元数据（由sender线程去获取）
  2. 序列化消息，找到哪个partition
  3. 写到缓存队列（batch 批处理，提升网络吞吐），发往同一个分区里的消息会有一个单独的队列
  4. sender线程从上述队列中读取数据，提交发送出去给kafka集群对应的broker
  5. 异步发送后，再处理回调函数中的逻辑

消费者启动流程

  1. 根据消费组group_id，计算出来 coordinator 在哪个broker
  2. 给coordinator 发送JOIN_GROUP 请求，第一个注册过去的就认为是 leader consumer
  3. 如果自己是leader consumer，制定消费方案。当然组内所有其他的消费者，也由coordinator告诉 leader consumer。
  4. 消费方案制定好后再发送 SYNC_GROUP 请求给 coordinator

消费者提交位移

消费者在消费时，提交位移要写到位移topic对应的partition，而提交位移也正是和 coordinator交互。 coordinator 和这个leader partition 在同一个broker。

消费者心跳

消费者心跳 和 coordinator交互。coordinator 来检测某个消费者是否心跳失效，进而决定是否开启rebalance等。

参考

  • 《Kafka核心源码解读》
  • 《Kafka权威指南》
  • 《图解Kafka》