概述

在Kafka集群中会有一个或多个broker，其中有一个broker会被选举为控制器（Kafka Controler），它负责管理整个集群中所有分区和副本的状态。当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责分区的重新分配。

这里主要讲控制器的功能与实现，优雅关闭部分会简单概括。

控制器的选举与故障恢复

控制器的选举

Kafka中的控制器选举工作依赖于ZooKeeper，成功竞选为控制器的broker会在ZooKeeper中创建/controller这个临时（EPHEMERAL）节点，此临时节点的内容参考如下：
（version":1,brokerid":0,“timestamp":”1529210278988”）
其中version在目前版本中固定为1，brokeria表示成为控制器的broker的ia编号，timestamp 表示竞选成为控制器时的时间戳。

在任意时刻，集群中有且仅有一个控制器。每个broker启动的时候会去尝试读取/controller节点的brokeria的值，如果读取到brokerid的值不为-1，则表示已经有其他broker节点成功竞选为控制器，所以当前broker就会放弃竞选：如果ZooKeeper中不存在/controller，或者这个节点中的数据异常，那么就会尝试去创建/controller节点。当前broker去创建节点的时候，也有可能其他broker同时去尝试创建这个节点，只有创建成功的那个broker才会成为控制器，而创建失败的broker竞选失败。每个broker都会在内存中保存当前控制器的brokerid值，这个值可以标识为activeControllerid。

ZooKeeper中还有一个与控制器有关的/controller_epoch节点，这个节点是持久（PERSISTENT）节点，节点中存放的是一个整型的controller_epoch值。controller_epoch用于记录控制器发生变更的次数，即记录当前的控制器是第几代控制器，我们也可以称之为“控制器的纪元”

controller_epch的初始值为1，即集群中第一个控制器的纪元为1，当控制器发生变更时，每选出一个新的控制器就将该字段值加1。每个和控制器交互的请求都会携带controller_epoch这个字段，如果请求的controller_epoch值小于内存中的controller_epoch值，则认为这个请求是向已经过期的控制器所发送的请求，那么这个请求会被认定为无效的请求。如果请求的controller_epoch值大于内存中的controller_epoch值，那么说明已经有新的控制器当选了。由此可见，Kafka通过controller_epoch来保证控制器的唯性，进而保证相关操作的一致性。

具备控制器身份的broker需要比其他普通的broker多一份职责，具体细节如下：

• 监听分区相关的变化。为ZooKeeper中的/admin/reassign partitions节点注 册
  PartitionReassignmentHandler，用来处理分区重分配的动作。为ZooKeeper中的/isr change notification节点注册IsrChangeNotificetionHandler，用来处理ISR集合变更的动作。为ZooKeeper中中的/admin/preferred-replica-election节点添加
  PreferredReplicaElectionHandler，用来处理优先副本的选举动作。

• 监听主题相关的变化。为ZooKeeper中的/brokers/topics节点添加
  TopicChangeHandler，用来处理主题增减的变化；为ZooKeeper中的/admin/ delete
  topics节点添加TopicDeletionHandler，月用来处理删除主题的动作。

• 监听broker相关的变化。为ZooKeeper中的/brokers/ids节点添加BrokerChangeHandler，用来处理broker增减的变化。

• 从ZooKeeper中读取获取当前所有与主题、分区及broker有关的信息并进行相应的管理。对所有主题对应的ZooKeeper中的/brokers/topics/节点添加PartitionModificationsHandler，用来监听主题中的分区分配变化。

• 启动并管理分区状态机和副本状态机。

• 更新集群的元数据信息。

• 如果参数auto.leader.rebalance.enable设置为true，贝则还会开启一个名为
  “auto-leader-rebalance-task”的定时任务来负责维护分区的优先副本的均衡。

控制器在选举成功之后会读取ZooKeeper中各个节点的数据来初始化上下文信息（ControllerContext），并且需要管理这些上下文信息。比如为某个主题增加了若干分区，控制器在负责创建这些分区的同时要更新上下文信息，并且需要将这些变更信息同步到其他普通的broker节点中。不管是监听器触发的事件，还是定时任务触发的事件，或者是其他事件都会读取或更新控制器中的上下文信息，那么这样就会涉及多线程间的同步。如果单纯使用锁机制来实现，那么整体的性能会大打折扣。针对这一现象，Kafka的控制器使用单线程基于事件队列的模型，将每个事件都做一层封装，然后按照事件发生的先后顺序暂存到LinkedBlockingQueue中，最后使用一个专用的线程（ControllerEventThread）按照FIFO（FirstInputFirstOutput，先入先出）的原则顺序处理各个事件，这样不需要锁机制就可以在多线程间维护线程安全，具体可以参考下图。

在Kafka的早期版本中，并没有采用KafkaController这样一个概念来对分区和副本的状态进行管理，而是依赖于ZooKeeper，每个broker都会在ZooKeeper上为分区和副本注册大量的监听器（Watcher）。当分区或副本状态变化时，会唤醒很多不必要的监听器，这种严重依赖

ZooKeeper的设计会有脑裂、羊群效应，以及造成ZooKeeper过载的隐患（旧版的消费者客户端存在同样的问题）。在目前的新版本的设计中，只有KafkaController在ZooKeeper上注册相应的监听器，其他的broker极少需要再监听ZooKeeper中的数据变化，这样省去了很多不必要的麻烦。不过每个broker还是会对/controller节点添加监听器，以此来监听此节点的数据变化（ControllerChangeHandler）

故障恢复

当/controller节点的数据发生变化时，每个broker都会更新自身内存中保存的activeControllerlds如果broker在数据变更前是控制器，在数据变更后自身的brokerid值与新的 activeControllerId值不一致，那么就需要“退位”，关闭相应的资源，比如关闭状态机、注销相应的监听器等。有可能控制器由于异常而下线，造成/controller这个临时节点被自动删除；也有可能是其他原因将此节点删除了。

当/controller节点被删除时，每个broker都会进行选举，如果broker在节点被删除前是控制器，那么在选举前还需要有一个“退位”的动作。如果有特殊需要，则可以手动删除/controller节点来触发新一轮的选举。当然关闭控制器所对应的broker，以及手动向/controller节点写入新的brokerid的所对应的数据，同样可以触发新一轮的选举。

优雅关闭

Kafka自身提供了一个脚本工具，就是存放在其bin目录下的kafka-server-stop.sh，这个脚本的内容非常简单，具体内容如下：

PIDS=S (ps ax I grep -i kafkal.Kafka’ Igrep java l grep -v grep 1 awk print S1）'） if[-Z”SPIDS”】；then echo rNokafkaserver to stop' exit 1 else kill -S TERM SPIDS fi

以上脚本直接执行会存在执行不成功，这是因为与Kafka进程有关的输出信息太长，所以kafka-server-stop.sh脚本在很多情况下并不会奏效。

可以通过修改计算机PAGE_SIZE的大小或者修改脚本内容，这里我们可以直接修改kafka-server-storsh脚本的内容，将其中的第一行命令修改如下：
PIDS=S(ps axIgrep -i 'kafka’i grep javalgrep-v greplawkitprint Si}'
即把“.Kafka”去掉，这样在绝大多数情况下是可以奏效的。如果有极端情况，即使这样也不能关闭，那么只需要按照以下两个步骤就可以优雅地关闭Kafka的服务进程：
（1）获取Kafka的服务进程号PIDS。可以使用Java中的jps命令或使用Linux系统中的ps命令来查看。
（2）使用kill-STERMSPIDS或kill-15 $PIDS的方式来关闭进程，注意千万不要使用kill-9的方式。

为什么这样关闭的方式会是优雅的？Kafka服务入口程序中有一个名为“kafka-shutdown-hock”的关闭钩子，待Kafka进程捕获终止信号的时候会执行这个关闭钩子中的内容，其中除了正常关闭一些必要的资源，还会执行一个控制关闭C ControlledShutdown ）的动作。使用ControlledShutdown的方式关闭Kafka有两个优点：一是可以让消息完全同步到磁盘上，在服务下次重新上线时不需要进行日志的恢复操作；二是ControllerShutdown在关闭服务之前，会对其上的leader副本进行证移，这样就可以减少分区的不可用时间。这里更加详细的分析可以查下其他人的博客。

当然这里是通过脚本进行优雅关闭，你也可以自己通过KafkaAdminClient自己写一个优雅关闭的接口去执行。这里详细的设计与实现可以查下其他博客。

分区leader的选举

分区leader副本的选举由控制器负责具体实施。当创建分区（创建主题或增加分区都有创建分区的动作）或分区上线（比如分区中原先的leader副本下线，此时分区需要选举一个新的leader上线来对外提供服务）的时候都需要执行leader的选举动作，对应的选举策略为OfflinePartitionLeaderElectionStrategy。这种策略的基本思路是按照AR集合中副本的顺序查找第一个存活的副本，并且这个副本在ISR集合中。一个分区的AR集合在分配的时候就被指定并且只要不发生重分配的情况，集合内部副本的顺序是保持不变的，而分区的ISR集合中副本的顺序可能会改变。注意这里是根据AR的顺序而不是ISR的顺序进行选举的。

如果ISR集合中没有可用的副本，那么此时还要再检查一下所配置的unclean.leader.election.enable参数（默认值为false）。如果这个参数配置为true那么表示允许从非ISR列表中的选举leader，从AR列表中找到第一个存活的副本即为leader。

那么哪些情况下会出现leader选举呢？

• 当分区进行重分配的时候也需要执行leader的选举动作，对应的选举策略为ReassignPartitionLeaderElectionStrategy。这个选举策略的思路比较简单：从重分配的AR列表中找到第一个存活的副本，且这个副本在日前的ISR列表中。

• 当发生优先副本的选举时，直接将优先副本设置为leader即可，AR集合中的第一个副本即为优先副本（PreferredReplicaPartitionLeaderElectionStrategy）

• 还有一种情况会发生leader的选举，当某节点被优雅地关闭（也就是执行 ControlledShutdown）时，位于这个节点上的leader副本都会下线，所以与此对应的分区需要执行leader的选举。与此对应的选举策略（ControlledShutdownPartitionLeaderElectionStrategy）为：从AR列表中找到第一个存活的副本，且这个副本在日前的ISR列表中，与此同时还要确保这个副本不处于正在被关闭的节点上。