浅谈Kafka Controller（控制器）的相关知识
前言
七夕快乐呀~
（看官如果想听一首应景的歌的话，可以移步上一篇hhhhhh
Kafka作为一个高效的分布式消息系统，在多处关键点都采用了主从（或者说Leader-Follower）的设计思路，例如：
- Broker主从设计，主节点称为Controller；
- Partition Replica主从设计，处理客户端请求的主要Replica称为Partition Leader；
- Consumer Group Rebalance过程中的Consumer主从设计，负责确定Partition分配规则的那个Consumer也称为Leader（详情可参见这篇文章）。
本文就来谈谈第一个，即Controller的一些细节。
Controller简介及选举
Controller（控制器）本质上就是Kafka集群里的一个Broker。
它除了负责普通Broker该做的事情（存储、发送、接收、处理消息）之外，还额外负责管理Kafka集群中所有Broker、Partition和Replica的状态。
每个Broker在启动时都会实例化KafkaController对象，而Controller的选举是各Broker通过在ZooKeeper中抢注/controller临时节点实现，第一个成功注册该节点的Broker就成为真正的Controller。其他Broker则在该节点上注册监听，如果当前Controller失败，/controller节点消失，就会触发其他Broker重新进行选举。
/controller节点中保存的信息如下。
[zk: localhost:2181(CONNECTED) 0] get /kafka/controller
{"version":1,"brokerid":123,"timestamp":"1589027593271"}
其中，brokerid字段表示Controller对应Broker的唯一标识，timestamp字段表示最近一次Controller发生变化的时间戳。
除了/controller之外，ZK中还有一个与Controller相关的持久节点/controller_epoch，保存有一个正整数，表示当前Controller的纪元值（即代数）。该值初始为1，每选举一次新的Controller就将它加1。
[zk: localhost:2181(CONNECTED) 1] get /kafka/controller_epoch
13
这个纪元值有什么用处呢？当客户端与Controller交互时，都会带上自己缓存的纪元值，如果请求中的纪元值比ZK里的纪元值小，说明该客户端试图与已过期的Controller交互，该请求会被视为无效。也就是说，Kafka通过Controller纪元值的唯一性保证一致性，防止集群脑裂（split brain）。另外，如果纪元值非常大，说明Kafka集群不稳定，总是重新选举Controller，需要特别注意。
Controller的具体作用
前文说到Controller“负责管理Kafka集群中所有Broker、Partition和Replica的状态”，还是太泛泛了。由于KafkaController类的代码很长，不适合逐段讲解，因此下面以文字形式列举Controller需要做些什么，共9项。
-
当Producer或Consumer通过MetadataRequest请求查询Partition元数据（如Leader和ISR信息）时，将发生变化的Partition元数据广播给各个Broker。
-
处理ControlledShudownRequest请求，该请求用于优雅地关闭一个Broker（主要是会主动删除ZK中对应的Broker节点，减少对应Partition不可用的时间）。
-
启动并管理Partition状态机组件（PartitionStateMachine）和Replica状态机组件（ReplicaStateMachine）。
-
注册TopicChangeListener监听器，监听ZK的
/brokers/topics节点，处理Topic的增删操作；注册TopicDeletionListener监听器，监听ZK的/admin/delete_topics节点，用来实际执行删除Topic的动作。 -
注册PartitionModificationsListener监听器，监听ZK中各个
/brokers/topics/<topic>节点，处理Topic的Partition扩容和缩容操作。 -
注册PreferredReplicaElectionListener监听器，监听ZK的
/admin/preferred_replica_election节点，处理最优Replica的重选举。 -
注册IsrChangeNotificetionListener监听器，监听ZK的
/isr_change_notification节点，处理ISR（in-sync replicas）集合发生变化的Partition。 -
注册PartitionReassignmentListener监听器，监听ZK的
/admin/reassign_partitions节点，用于重新分配各Partition的Leader和Follower。 -
注册BrokerChangeListener监听器，监听ZK的
/brokers/ids节点，触发Broker的上下线操作。
可见，Controller很大程度上是通过ZK来发挥作用的。如果看官对Kafka在ZK中维护的数据结构不熟悉，可以参见笔者之前写的这篇文章。
Controller架构概览
最后来看看Controller的架构简图，如下所示。注意Controller在0.11版本经历了一次比较大的重构，这里是重构之后的设计。
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Controller+Redesign
由图可见，Controller主要由以下5部分组成。
-
ZK监听器：已经说过了，不再废话。
-
定时任务：举个例子，假设我们将
auto.leader.rebalance.enable参数设为true，那么就会启动名为auto-leader-rebalance-task的定时任务来自动维护最优Replica的平衡。 -
Controller上下文：在Controller初始化阶段，从ZK中已存储的数据建立，并在Controller的生命周期中一直维护。包含集群Broker可达性信息，与所有Topic、Partition、Replica的状态信息。
-
事件队列：本质上为FIFO的阻塞队列（LinkedBlockingQueue），承载各个监听器、定时任务投递过来的状态变更信息，这些信息都包装为事件。
-
事件处理线程：顾名思义，只有单线程，用来处理各个事件，并将它们的结果反映到Controller上下文，以及异步地propagate到各个Broker中。使用单线程的好处是无需关心多线程的同步，无锁机制可以提升性能。
The End
今晚忙着过节，空闲时间不甚多，随手写了几笔，看官随意看看就好。
民那晚安晚安。
