无镜--kafka之控制器（一）

在kafka中每个消息代理节点（broker）都管理着集群中所有分区的一部分。kafka的控制器（KafkaController）的工作有很多：分区的分配；分区主副本的选举；代理节点启动或者下线；处理代理节点的故障转移；创建，删除主题；新增分区时，处理分区的重新分配；管理所有分区的状态机和副本的状态机；处理状态机的变化事件。

kafka集群的一些信息记录到ZK中：集群中所有的代理节点，topic的所有分区，分区的副本信息（副本集，主副本，ISR副本集）。外部事件会更新ZK的数据，ZK数据一旦发生变化，KafkaController都要做出不同的响应处理。

外部事件，会更新ZK的不同节点：

1，节点上线或者下线，更新ZK的/brokers/ids节点。

2，创建或删除主题，更新ZK的/brokers/topics节点。

3，增加或减少分区数，更新ZK中与主题相关的节点。比如：/brokers/topics/[topic]。

4，增加或减少副本数，更新ZK中与分区相关的节点。比如：/brokers/topics/[topic]/[partition_id]。

控制器

kafka服务端每个代理节点都有一个控制器对象。和副本一样，控制器也会选择出一个主控制器。利用ZK的领导选举机制，只有一个代理节点能成为主控制器，其他代理节点只有在主控制器出现故障或者会话失效时才参与领导选举：每个代理节点都会成为ZK的客户端，向ZK服务端尝试创建/controller临时节点，但最终只有一个代理节点能成功创建/controller临时节点。由于主控制器创建的ZK节点是临时节点，因此当主控制器出现故障或者会话失效时，临时节点会被删除，这个时候所有的代理节点都会尝试创建/controller临时节点，一个创建成功后成为集群新的主控制器节点。

每个代理节点都需要和ZK交互，作为ZK的客户端，建立和ZK服务端的网络连接。在启动控制器时，会先注册一个会话失效的监听器，然后控制器通过基于ZK的选举器启动选举过程。ZK客户端注册监昕器，实际上是注册了ZK的Watcher 。ZK的Watcher都是一次性的，当会话失效后，客户端除了会重新创建临时节点选举新的主控制器，还需要再次注册会话失效的监昕器。

每个控制器都有一个选举器，并且都会注册一个会话失效的监控器，都会注册一个数据改变的监听器。具体步骤：

1，假设有三个代理节点，它们启动时都会通过创建/controller节点竞选主控制器，假设代理节点3创建/controller节点成功，成为主控制器。

2，三个代理节点都会注册会话失效监听器，都会在/controller节点注册数据改变监听器。

3，第一个代理节点的会话失效了，它的选举器尝试重新创建/controller节点，但是会创建失败。

4，第二个代理节点的会话失效了和第一个代理节点一样。

5，第三个代理节点的会话失效了，由于创建的是临时节点，会话失效时，/controller节点会被删除。这时三个代理节点都会收到数据改变的事件，它们的选举器都会尝试重新创建/controller节点。重新竞选主控制器。

来之《Kafka技术内幕：图文详解Kafka源码设计与实现》： 基于ZK选举器选举主控制器流程

控制器上下文

主控制器会读取ZK中的集群元数据，构造出控制器的上下文对象（ControllerContext）。

上下文对象中会保存主题信息；分区信息；初始化控制器的通道管理器，建立到集群各个代理节点的网络连接；初始化“选举最优副本作为主副本”，“重新分配分区”，“删除主题的管理器”。这样上下文对象是控制器工作时数据存储的介质。控制器的具体动作主要通过事件的监听器触发，上下文和事件监听器都与ZK相关，上下文从ZK读取数据，事件监听器注册监听器到ZK节点。

1，控制器向ZK节点注册各种监听器，每种监听器都有具体的事件处理逻辑。

2，更新ZK节点的数据，触发监听器调用不同的回调方法。

3，控制器执行具体的事件处理逻辑，处理完成后，再次注册监听器，为下次事件处理做准备。

上下文对象保存了上一次事件发生时的数据，而ZK保存了最新的数据，两者的数据会不一致。与ZK节点关联的监听器触发事件处理时，监听器要对比ZK节点和上下文对象中的数据，找出新增和删除的数据，具体步骤如下 。

1，ZK节点的最新数据减去控制器上下文数据表示需要新增的数据 。

2，控制器上下文数据减去ZK节点的最新数据表示需要删除的数据 。

3，将控制器上下文数据更新为ZK节点的最新数据。

4，让控制器分别处理需要新增和删除数据对应的事件。

ZK监听器

主题改变的监听器：监听/brokers/topics/的子节点变化事件。当主题发生变化时，监听器会处理主题的增加和删除事件。比如创建主题时，kafka会往ZK节点/brokers/topics/添加子节点/brokers/topics/[topic_name]，并触发监听器调用相关方法。

分区改变的监听器：监听/brokers/topics/[topic]节点的数据变化事件。当主题的分区发生变化时，监听器会处理分区相关的事件。增加分区时，kafka会修改ZK节点/brokers/topics/[topic_name]的数据内容。对于topic中新增的分区，监听器会调用相关方法创建新的分区。

代理节点改变的监听器：监听/brokers/ids的子节点变化事件。当代理节点发生变化时，监听器会处理代理节点的上线和下线事件。比如代理节点宕机，kafka会删除/brokers/ids/[broker_id]子节点，并触发监听器调用相关方法；代理节点上线，kafka会创建/brokers/ids/[broker_id]子节点，并触发监听器调用相关方法。

参考资料：

Kafka技术内幕：图文详解Kafka源码设计与实现