学习方法｜一把解开Kafka背后机制的“钥匙”

发表于 2年以前 | 总阅读数：326 次

在Kafka中有一个非常重要的角色：控制器，KafkaController，承担着Kafka核心运作机制，从本文开始将逐步深入Kafka内核，揭晓Kafka内部的运作机制，为更好的运维Kafka储备充足的弹药。

深入探究Kafka内核，我们将如何展开呢？经过笔者初略的浏览了KafkaController类的源码，发现Kafka的控制器严重依赖Zookeeper集群，而基于Zookeeper编程的常规套路：基于事件监听模型：

Kafka Zookeeper事件监听如下图所示：

基于此种编程模型，事件回调方法就成为我们探究Kafka的突破口，而kafka也为控制类事件定义了统一的接口:ControllerEvent,简单声明如下图所示：

关键信息如下：

ControllerEvent的所有子类如下图所示：

上面每一种事件都代表Kafka内核蕴含的一种机制，这些也是接下来探究的重点，本文将重点关注Startup事件，包含了Kafka控制器选举，揭晓控制器的职权范围。

1、Kafka控制选举机制

通过查看Kafka启动流程发现，Broker启动时首先会向/broker/ids节点下注册自己，然后会将Startup事件放入到任务队列中，从而将触发Controller的选举，Startup事件的定义如下图所示：

主要完成两件事情：

接下来将详细探究其实现细节，将其运作机制掌握于心。

Kafka Controller的选举由具体由KafkaController的elect()方法实现，接下来将对该方法进行详细剖析。

核心实现步骤如下：；

尝试从zookeeper中查看 /controller 节点中的内容，如果返回的值不等与-1，表示集群已经选举出KafkaController，则无需再次选举，直接返回。
发起KafkaController的选举，具体做法是尝试向zookeeper创建 /controller 节点，该行为会原子的返回如下两种结果：
节点已经存在如果节点已经存在，则说明其他Broker提前创建了该节点，也就是获得Controller选举的胜利，如果节点存在，则出发异常处理(即当前节点没有成功选举后的善后操作，稍后会详细解读)
节点成功创建如果节点创建成功，则自己在本轮Controller选举中取得胜利，成功成为新的kafka Controller。
如果节点成功被选举为新的Controller，则触发Controller故障转移的相关逻辑。
如果选举失败，则进行必要的善后工作，接下来探究。

当一个新节点被选举成Kafka Controller后，也就是“新的秩序”重新建立，需要做哪些事情呢？由KafkaController的onControllerFailover方法实现，该方法代码有点多，将分步骤介绍。

Step1：重新注册ChildChange事件，这类事件主要是关注其子节点的新增/删除/修改时间，代码如下图所示：

详情说明如下：

/brokers/ids 每一个子节点代表一个Broker，监听broker的加入与下线并做出对应的响应
/brokers/topics 记录所有主题的元信息，主要是分区信息(分区数量，各个分区的leader节点、ISR集合等)。
/admin/delete_topics 待删除的topic。
/log_dir_event_notification 如果当某一个Broker节点的日志读写出现异常，例如磁盘损坏，会向zookeeper，从而触发副本下线机制，触发相应的分区Leader选举
/isr_change_notification ISR变更通知

上述节点的事件监听，背后都蕴含了Kafka相关的机制，后续会一一揭晓。

Step2：重新注册Node节点改变事件，代码如下图所示：

具体订阅的节点如下：

上述节点的事件监听，同样背后都蕴含了Kafka的相关机制，这些后续都会一一介绍，从而揭晓Kafka核心机制。

Step3： 删除/log_dir_event_notification、/isr_change_notification相关的子节点，新的Controller节点选举成功后，分区故障转移，ISR变更都将忽略，这些事项后续会重新触发，代码如下图所示：

Step4：初始化控制器上下文，核心代码如下图所示：

Kafka Controller的上下文主要包括如下信息：

获取当前所有活跃的Broker节点信息(获取 /brokers/ids 节点下的子节点)
获取集群中所有的Topic元数据，后续会将这些信息传播到各个Broker节点，也正是基于如此，生产者、消费者无需再连接Zookeeper。
为每一个主题创建Watch，监听主题的元信息变更事件。
获取每一个主题的分区分布情况，即每一个分区Leader所在的brokerId，副本在哪些Broker中，/broker/topics/{topicName}节点中存储的数据如下图所示：
为每一个/broker/ids/{id} broker注册节点内容变更事件，即可以监听broker元数据的变更，其实主要是监听endpoints的变化，如果发生变化，KafkaController会向所有注册(包括被认为正在关闭中的broker)发送UpdateMetadata命令。
从zk中加载所有分区的ISR与Leader Epoch,我们可以看看一个分区的状态：
启动控制器的通道管理器ControllerChannelManager，管理着控制器与集群内其他Broker中的控制通道，并且包含QueueSize、RequestRateAndQueueTimeMs等重要监控指标，这部分后续会深入展开。
读取/admin/reassign_partitions中的内容(Broker端分区重新分布计划)，触发PartitionReassignmentIsrChangeHandler事件，关于分区重新分布的运维实战可以查阅笔者：https://mp.weixin.qq.com/s/X4zmnIg0zP3XFnogC45k4w

Step5： 获取需要删除与不需要删除的topic，代码如下图所示：

具体的处理逻辑：

首先从zookeeper中读取到 /kafka_cluster_01/admin/delete_topics节点下的所有topic，执行topic删除命令时首先会往该节点上添加一个节点。
这些节点中如果topic的分区的状态为online(在线)或者分区在执行分区移动，则这些topic将不会立即被删除，存入topicsIneligibleForDeletion集合中。

Step6：启动副本状态机与分区状态机，代码如下图所示：

Kafka中分区、副本是存储的核心，分区、副本各自有多种状态，在Kafka内部实现中使用状态机来实现，其中副本的状态机状态驱动示意图如下所示：

分区的状态比较简单，其状态驱动如下图所示：

Step7：处理分区重分配、topic删除、触发倾向性副本的选举，具体代码如下图所示：

核心要点：

如果/admin/reassign_partitions中节点存储了topic的分区的重分配计划，则控制器选举成功后需要触发该部分逻辑的处理。
如果Broker端允许删除topic，配置项(delete.topic.enable),默认为true，如果/admin/delete_topics节点存在子节点，并且需要执行删除动作，则触发主题的删除动作。
如果/admin/preferred_replica_election节点存储了需要进行倾向性选举，则触发倾向性选举。