Kafka——Controller设计详解

谁借莪１个温暖的怀抱￠ 2023-10-14 21:03 19阅读 0赞

#### 这里写目录标题 ####

*  Controller设计
 *   *  1、Controller概览
     *  2、Controller管理状态
     *   *  2.1、副本状态机(Replica State Machine)
         *  2.2、分区状态机(Partition State Machine)
     *  3、Controller职责
     *   *  3.1、更新集群元数据
         *  3.2、创建topic
         *  3.3、删除topic
         *  3.4、分区重分配
         *  3.5、preferred leader选举
         *  3.6、topic分区扩展
         *  3.7、broker加入集群
         *  3.8、broker崩溃
         *  3.9、受控关闭
         *  3.10、controller leader选举
     *  4、controller与broker间的通信
     *  5、controller组件
     *   *  5.1、数据类组件——ControllerContext
         *  5.2、基础功能类
         *  5.3、状态机类
         *  5.4、选举器类
         *  5.5、Zookeeper监听器

## Controller设计 ##

### 1、Controller概览 ###

在一个Kafka集群中，某个broker会被选举出来承担特殊的角色，即控制器（下称controller)。顾名思义，引入controller就是用来管理和协调Kafka集群的。具体来说，就是管理集群中所有分区的状态并执行相应的管理操作。

每个Kafka集群任意时刻都只能有一个controller。当集群启动时，所有broker都会参与controller的竞选，但最终只能由一个broker胜出。一旦controller在某个时刻崩溃，集群中剩余的broker会立刻得到通知，然后开启新一轮的controller选举。新选举出来的controller将承担起之前controller的所有工作。下图展示了controller的架构。

![在这里插入图片描述][651b6af09c894f34af0210d769df93cf.png]

### 2、Controller管理状态 ###

简单地说，controller维护的状态分为两类：每台broker上的分区副本和每个分区的leader副本信息。从维度上看，这些状态又可分为副本状态和分区状态。controller为了维护这两个状态专门引入了两个状态机，分别管理副本状态和分区状态。

#### 2.1、副本状态机(Replica State Machine) ####

当前，Kafka为副本定义了7种状态以及每个状态之间的流转规则。这些状态分别如下。

*  `NewReplicac`:ontroller创建副本时的最初状态。当处在这个状态时，副本只能成为follower副本。
 *  `OnlineReplica`:启动副本后变更为该状态。在该状态下，副本既可以成为follower副本也可以成为leader副本。
 *  `OfflineReplica`:一旦副本所在broker崩溃，该副本将变更为该状态。
 *  `ReplicaDeletionStarted`1:若开启了topic删除操作，topic下所有分区的所有副本都会被删除。此时副本进入该状态。
 *  `ReplicaDeletionSuccessful`:若副本成功响应了删除副本请求，则进入该状态。
 *  `ReplicaDeletionIneligible`:若副本删除失败，则进入该状态。
 *  `NonExistentReplica`:若副本被成功删除，则进入该状态。

副本状态机的流转规则如图所示。当创建某个topic后，该topic下所有分区的所有副本都是NonExistent状态的，此时controller加载ZooKeeper中该topic每个分区的所有副本  
信息到内存中，同时将副本状态变更为New,之后controller选择该分区副本列表中的第一个副本作为分区的leader副本并设置所有副本进入ISR,然后在ZooKeeper中持久化该决定。

一旦确定了分区的leader和ISR之后，controller会将这些信息以请求的方式发送给所有副本，同时将这些副本状态同步到集群的所有broker上以便让它们知晓。当这些都做完后，controller会将分区的所有副本状态置为Online,这也是副本正常工作的状态。

当开启了topic删除操作时，controller会尝试停止所有副本，此时副本将停止向leader获取数据，但若停止的副本就是leader副本本身，则controller会设置该分区的leader为  
NO LEADER,之后副本进入Offline状态。一旦所有副本进入Offline状态，controller需要将副本进一步变更到ReplicaDeletionStarted状态表明删除topic任务的开启。在这一步的状态流转中，controller会给这些副本所在的broker发送请求，让它们删除本机上的副本数据。一  
旦删除成功，这些副本就变更到ReplicaDeletionSuccessful状态；如果有失败的副本，那么该副本进入ReplicaDeletionIneligible状态表明暂时还无法删除该副本，等待controller的重试。

那些处于ReplicaDeletionSuccessful状态的副本稍后会被自动地变更到NonExistent终止状态，同时controller的上下文缓存会清除这些副本信息。这就是副本状态机操作副本状态流转的典型场景。

![在这里插入图片描述][7e92dfd1e98349009bbb5a924ccda0d0.png]

#### 2.2、分区状态机(Partition State Machine) ####

除了副本状态机，controller还引入了分区状态机来负责集群下所有分区的状态管理：

分区状态机只定义了4个分区状态，它们分别如下。

*  `NonExistent`:表明不存在的分区或已删除的分区。
 *  `NewPartition`:一旦被创建，分区便处于该状态。此时，Kafka己经为分区确定了副本列表，但尚未选举出leader和ISR。
 *  `OnlinePartition`:一旦该分区的leader被选出，则进入此状态。这也是分区正常工作时的状态。
 *  `OfflinePartition`:在成功选举出leader后，若leader所在的broker宕机，则分区将进入该状态，表明无法正常工作了。

![在这里插入图片描述][6a39eb4c3a284356821562906b7d493e.png]

比起副本状态机7个状态，分区状态机的状态空间要小得多，分析它们之间状态的流转也要容易得多。当创建topic时，controller负责创建分区对象，它首先会短暂地将所有分区状态设置为NonExistent,之后马上读取ZooKeeper中（严格来说是从controller的上下文信息中读取，关于controller的上下文信息稍后会详细讨论)的副本分配方案，然后令分区状态进入NewPartition。处于NewPartition状态的分区尚未有leader和ISR,因此controller会初始化leader和ISR信息并设置分区状态为OnlinePartition。此时分区开始正常工作。

若用户发起了删除topic操作或关闭broker操作，那么controller会令受影响的分区进入到Offline状态。如果是删除操作，则controller还会开启分区下面副本的删除操作并最终将分区状态设置为NonExistent。而如果是关闭broker操作，则controller会判断该broker是否是分区的leader。如果是则需要开启新一轮的leader选举并调整分区状态回OnlinePartition。

### 3、Controller职责 ###

对集群状态的维护只是controller保持运行状态一致性的一个基本要素，但却不是controller的职责所在。应该这样说，如果保持controller持续稳定地对外提供服务，就必须要求controller妥善地保存这些状态。实际上，controller的职责相当多，包括如下职责。

*  更新集群元数据信息。
 *  创建topice
 *  删除topic.
 *  分区重分配。
 *  preferred leader副本选举。
 *  topic分区扩展。
 *  broker加入集群。
 *  broker崩溃。
 *  受控关闭。
 *  controller leader选举。

#### 3.1、更新集群元数据 ####

一个clients能够向集群中任意一台broker发送METADATA请求来查询topic的分区信息(比如topic有多少个分区、每个分区的leader在哪台broker上以及分区的副本列表)。随着  
集群的运行，这部分信息可能会发生变化，因此就需要controller提供一种机制，用于随时随地地把变更后的分区信息广播出去，同步给集群上所有的broker。具体做法就是，当有分区信息发生变更时，controller将变更后的信息封装进UpdateMetadataRequests请求（通信协议中的  
一种)中，然后发送给集群中的每个broker,这样clients在请求数据时总是能够获取最新、最及时的分区信息。

#### 3.2、创建topic ####

controller启动时会创建一个ZooKeeper的监听器，该监听器的唯一任务就是监控ZooKeeper节点brokers/topics下子节点的变更情况。当前一个clients或admin创建topic的方  
式主要有如下3种。

*  通过kafka-topics脚本的-create创建。
 *  构造CreateTopicsRequest请求创建。
 *  配置broker端参数auto.create.topics.enable为true,然后发送MetadataRequest请求。

无论上述哪种方式，其基本的原理都是在ZooKeeper的brokers/topics下创建一个对应的znode,然后把这个topic的分区以及对应的副本列表写入这个znode中。之前所说的controller监听器一旦监控到该目录下有新增znode,就立即触发topic创建逻辑，即controller会为新建topic的每个分区确定leader和ISR,然后更新集群的元数据信息。做完这些之后controller还会创建一个新的监听器用于监听ZooKeeper的brokers/topics/<新增topic>节点内容的变更。这样当topic分区发生变化的时候controller也能在第一时间得到通知。

#### 3.3、删除topic ####

标准的Kafka删除topic方法有如下两种。

*  通过kafka-topics脚本的-delete来删除topic。
 *  构造DeleteTopicsRequest.。

这两种方式都是向ZooKeeper的/admin/delete topics下新建一个znode。controller启动时会创建一个监听器专门监听该路径下的子节点变更情况，一旦发现有新增节点，则controller立即开启删除topc的逻辑。该逻辑主要分为两个阶段：①停止所有副本运行；②删除所有副本的日志数据。一旦完成这些操作，controller会移除/admin/delete topics/<待删除topic>节点，这代表topic删除操作正式完成。

#### 3.4、分区重分配 ####

分区重分配操作通常都是由Kafka集群的管理员发起的，旨在对topic的所有分区重新分配副本所在broker的位置，以期望实现更均匀的分配效果。在该操作中管理员需要手动制定分配方案并按照指定的格式写入ZooKeeper的/admin/reassign\_partitions节点下。

分区副本重分配的过程实际上是先扩展再收缩的过程。controller首先将分区副本集合进行扩展(I旧副本集合与新副本集合的合集)，等待它们全部与leader保持同步之后将leader设置为新分配方案中的副本，最后执行收缩阶段，将分区副本集合缩减成分配方案中的副本集合。

#### 3.5、preferred leader选举 ####

为了避免分区副本分配不均匀，Kafka引入了preferred副本的概念。比如一个分区的副本列表是\[l,2,3\],那么broker1就被称为该分区的preferred leader,因为它位于副本列表的第一位。在集群运行的过程中，分区的leader因为各种各样的原因会发生变更，从而使得leader不再是preferred leader,此时用户可以发起命令将这些分区的leader重新调整为preferred leader。.具体的方法有如下两种。

*  设置broker端参数auto.leader.rebalance.enable为true,这样controller会定时地自动调整preferred leader。.
 *  通过kafka-preferred-replica-election脚本手动触发。

在内部，这两种方法都向ZooKeeper的/admin/preferred replica\_election节点写入数据。同样地，controller也会注册监听该目录的监听器。一旦被触发，controller会将对应分区的leader调整回副本列表中的第一个副本，之后将此变更广播出去。

#### 3.6、topic分区扩展 ####

在Kafka集群的运行过程中，用户可能会发现某些topic的现有分区数不足以支撑clients的业务量，因此可能有增加分区的需求。当前增加分区主要使用kafka-topics脚本的-alter选项来完成。和创建topic一样它会向ZooKeeper的brokers/topics//节点下写入新的分区目录。

前面说过，controller在创建topic之后会注册一个新的监听器用于监听分区目录数据的变化。因此一旦增加了topc分区，该监听器会被触发，执行对应的分区创建任务（比如选举leader和ISR等)，之后更新集群元数据信息。

#### 3.7、broker加入集群 ####

每个broker成功启动之后都会在ZooKeeper的broker/ids下创建一个znode,并写入broker的信息。如果要让Kafka动态地维护broker列表，就必须注册一个ZooKeeper监听器时刻监控该目录下的数据变化。

每当有新broker加入集群时，该监听器会感知到变化，执行对应的broker启动任务，之后更新集群元数据信息并广而告之。

#### 3.8、broker崩溃 ####

由于当前broker在ZooKeeper中注册的znode是临时节点，因此一旦broker崩溃，broker与ZooKeeper的会话会失效并导致临时节点被删除，故上面监控broker加入的那个监听器同样被用来监控那些因为崩溃而退出集群的broker列表。若发现有broker子目录“消失”，

controller便立即可知该broker退出集群，从而开启broker退出逻辑，最后更新集群元数据并同步到其他broker上。

#### 3.9、受控关闭 ####

“优雅”地关闭broker是指通过kafka-server-stop脚本、kill-9或kil-l5的方式关闭Kafka broker,而broker崩溃或强制退出通常都是以“掉电”或kill-9的方式实现的。前者被称为受控关闭。受控关闭能够最大限度地降低broker的不一致性。与其他controller功能中controller向其他broker发送请求不同的是，受控关闭是由即将关闭的broker向controller发送  
请求的。请求的名字是ControlledShutdownRequest。一旦发送完ControlledShutdownRequest,待关闭broker将一直处于阻塞状态，直到接收到broker端发送的ControlledShutdownResponse,表示关闭成功，或用完所有重试机会后强行退出。

controller在处理完必要的leader重选举和ISR收缩调整之后，会给broker发送ControlledShutdownResponse表明该broker现在可以正常退出。  
之前所有controller的功能都是依托于ZooKeeper的帮助来实现的，具体来说就是依靠ZooKeeper的监听器功能实现的。但对于受控关闭而言，它依赖于broker端的RPC来实现，即broker直接发送请求给controller,而没有借助ZooKeeper。

#### 3.10、controller leader选举 ####

作为Kafka集群的重要组件，controller必然要支持故障转移（fail-over)。若当前controller发生故障或显式关闭，Kafka必须要能够保证及时选出新的controller。当前，一个  
Kafka集群中发生controller leader选举的场景共有如下4种。

*  关闭controller所在broker。
 *  当前controller所在broker宕机或崩溃。
 *  手动删除ZooKeeper的/controller节点。
 *  手动向ZooKeeper的/controller节点写入新的broker id。

在用户没有完全理解controller的选举机制之前，不推荐用户使用后两种操作来触发controller的leader选举。  
万变不离其宗的是，这4种操作变更实际上都是/controller节点的内容，因此controller只需要做一件事情：创建一个监听该目录的监听器。/controller本质上是一个临时节点，节点保存了当前controller所在的broker id。集群首次启动时所有broker都会抢着创建该节点，但ZooKeeper保证了最终只能有一个broker胜出一胜出的那个broker即成为controller。

一旦成为controller,它会增加controller的版本号，即更新/controller epoch节点的值，然后履行上面所有的这些职责。对于那些没有成为controller的broker们而言，它们不会甘心失败，而是继续监听controller节点的存活情况并随时准备竞选新的controller。

### 4、controller与broker间的通信 ###

目前，controller启动时会为集群中所有broker创建一个专属的Socket连接（也包括controller所在的broker)。这就是说，若一个Kafka集群有l00台broker机器，那么controller  
会创建l00个Socket连接。虽说当前比较新的Kafka版本已经统一使用基于Java NIO Selector的网络连接库来实现这一功能，但controller依然会为每个TCP连接创建一个  
RequestSendThread线程。还是拿刚才的例子进行说明，l00台broker会创建l00个Socket连接和100个I/O线程。  
这些连接和线程被用于让controller给集群broker发送请求。在当前的设计下，controller只能给broker发送3种请求，它们分别如下。

*  `UpdateMetadataRequest`:之前反复提到的更新集群元数据请求。该请求携带了集群当前最新的元数据信息。接收到该请求后，broker会更新本地内存中的缓存信息，从而保证返还给clients的信息总是最新、最及时的。
 *  `LeaderAndIsrRequest`:用于创建分区、副本，同时完成作为leader和作为follower角色各自的逻辑。
 *  `StopReplicaRequest`:停止指定副本的数据请求操作，另外还负责删除副本数据的功能。

controller通常都是发送请求给broker的，但在controller职责部分中我们提到的ControlledShutdownRequest请求是例外，该请求是待关闭broker通过RPC直接发给controller  
的，即请求的流向是反的。另外这个请求还有一个特别之处，controller绝大多数的功能都是依托ZooKeeper来完成的，但只有这个请求是broker与controller直接进行交互完成的。

### 5、controller组件 ###

![在这里插入图片描述][a1d1116febee4f2d8e9cead117a703fc.png]

#### 5.1、数据类组件——ControllerContext ####

ControllerContext被称为controller上下文或controller缓存，可以说是controller最重要的数据组件了。它汇总了ZooKeeper中关于Kafka集群的所有元数据信息，是controller能够正确提供服务的基础。在0.1l.0.0版本之前，Kafka controller的设计是多线程的，因此保护好这个上下文，使其免受多线程并发修改就成了controller很重要的任务。事实上，0.11.0.0之前的版本中，controller被用户广为诟病的主要原因就是，使用了大量的同步机制来保护这个东西。

ControllerContext里面的内容非常丰富，如图所示。  
![在这里插入图片描述][1c2bbf062bd8430ca0cee8d46b6c0df6.png]

#### 5.2、基础功能类 ####

*  `ZkClient`:封装与ZooKeeper的各种交互API。controller与ZooKeeper的所有操作都交由该组件完成。
 *  `ControllerChannelManager`:前面说过了controller需要向其他broker发送请求，这个通道管理器即负责此事。
 *  `ControllerBrokerRequestBatch`:controller将发往同一broker的各种请求按照类型分组，稍后统一发送以提升效率。此组件就是用来管理请求batch的。
 *  `RequestSendThread`:负责给其他broker发送请求的I/O线程。
 *  `ZookeeperLeaderElector`:结合ZooKeeper负责controller的leader选举。

#### 5.3、状态机类 ####

*  `ReplicaStateMachine`:副本状态机，负责定义副本状态以及合法的副本状态流转。
 *  `PartitionStateMachine`:分区状态机，负责定义分区状态以及合法的分区状态流转。
 *  `TopicDeletionManager`:topic删除状态机，处理删除topic的各种状态流转以及相应的状态变更。

#### 5.4、选举器类 ####

当前，controller提供了4个选举器用于各种情况下的leader选举。注意，此处的leader选举是为分区进行leader选举而非选举controller。它们分别如下。

*  `OfflinePartitionLeaderSelector`:负责常规性的分区leader选举。
 *  `ReassignPartitionLeaderSelector`:负责用户发起分区重分配时的leader选举。
 *  `PreferredReplicaPartitionLeaderSelector`:负责用户发起preferred leader选举时的leader选举。
 *  `ControlledShutdownLeaderSelector`:负责broker在受控关闭后的leader选举。

#### 5.5、Zookeeper监听器 ####

controller自己维护的监听器实际上只有3个，它们分别如下。

*  `PartitionsReassignedListener`:监听ZooKeeper下分区重分配路径的数据变更情况。
 *  `PreferredReplicaElectionListener`:监听ZooKeeper下preferred leader选举路径的数据变更。
 *  `IsrChangeNotificationListener`:监听ZooKeeper下ISR列表变更通知路径下的数据变化。

这里简单解释一下上面最后一个监听器的含义。Kafka一旦发现topic分区的ISR发生了变化，就会在ZooKeeper的/isr change notification节点下写入一个新的数据节点，里面封装了集群中哪些topic的哪些分区对应的IS发生了变更。该监听器监控到节点变化后会发起“更新元数据”请求给集群中的所有broker。

事实上，controller定义的监听器远不止这3个，上面3个监听器只是controller自己维护的。更多的监听器交由各个状态机分别来维护。

[651b6af09c894f34af0210d769df93cf.png]: https://img-blog.csdnimg.cn/651b6af09c894f34af0210d769df93cf.png
[7e92dfd1e98349009bbb5a924ccda0d0.png]: https://img-blog.csdnimg.cn/7e92dfd1e98349009bbb5a924ccda0d0.png
[6a39eb4c3a284356821562906b7d493e.png]: https://img-blog.csdnimg.cn/6a39eb4c3a284356821562906b7d493e.png
[a1d1116febee4f2d8e9cead117a703fc.png]: https://img-blog.csdnimg.cn/a1d1116febee4f2d8e9cead117a703fc.png
[1c2bbf062bd8430ca0cee8d46b6c0df6.png]: https://img-blog.csdnimg.cn/1c2bbf062bd8430ca0cee8d46b6c0df6.png