xinyin025/CategoryResourceRepost
1
0
Fork 0
mirror of https://github.com/cheetahlou/CategoryResourceRepost.git synced 2025-11-16 06:03:45 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

mod

Browse Source
This commit is contained in:
louzefeng
2024-07-11 05:50:32 +00:00
parent bf99793fd0
commit d3828a7aee
6071 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
极客时间专栏/Kafka核心源码解读/Controller模块/11 | Controller元数据：Controller都保存有哪些东西？有几种状态？.md Normal file
View File

@@ -0,0 +1,287 @@
<audio id="audio" title="11 | Controller元数据：Controller都保存有哪些东西？有几种状态？" controls="" preload="none"><source id="mp3" src="https://static001.geekbang.org/resource/audio/5f/8b/5f0417c8727827798cb3225eba50f98b.mp3"></audio>
你好，我是胡夕。从今天开始，我们正式进入到第三大模块的学习：控制器（Controller）模块 。
提起Kafka中的Controller组件，我相信你一定不陌生。从某种意义上说，它是Kafka最核心的组件。一方面，它要为集群中的所有主题分区选举领导者副本；另一方面，它还承载着集群的全部元数据信息，并负责将这些元数据信息同步到其他Broker上。既然我们是Kafka源码解读课，那就绝对不能错过这么重量级的组件。
我画了一张图片，希望借助它帮你建立起对这个模块的整体认知。今天，我们先学习下Controller元数据。
<img src="https://static001.geekbang.org/resource/image/13/5f/13c0d8b3f52c295c70c71a154dae185f.jpg" alt="">
## 案例分享
在正式学习源码之前，我想向你分享一个真实的案例。
在我们公司的Kafka集群环境上，曾经出现了一个比较“诡异”的问题：某些核心业务的主题分区一直处于“不可用”状态。
通过使用“kafka-topics”命令查询，我们发现，这些分区的Leader显示是-1。之前，这些Leader所在的Broker机器因为负载高宕机了，当Broker重启回来后，Controller竟然无法成功地为这些分区选举Leader，因此，它们一直处于“不可用”状态。
由于是生产环境，我们的当务之急是马上恢复受损分区，然后才能调研问题的原因。有人提出，重启这些分区旧Leader所在的所有Broker机器——这很容易想到，毕竟“重启大法”一直很好用。但是，这一次竟然没有任何作用。
之后，有人建议升级重启大法，即重启集群的所有Broker——这在当时是不能接受的。且不说有很多业务依然在运行着，单是重启Kafka集群本身，就是一件非常缺乏计划性的事情。毕竟，生产环境怎么能随意重启呢？！
后来，我突然想到了Controller组件中重新选举Controller的代码。一旦Controller被选举出来，它就会向所有Broker更新集群元数据，也就是说，会“重刷”这些分区的状态。
那么问题来了，我们如何在避免重启集群的情况下，干掉已有Controller并执行新的Controller选举呢？答案就在源码中的**ControllerZNode.path**上，也就是ZooKeeper的/controller节点。倘若我们手动删除了/controller节点，Kafka集群就会触发Controller选举。于是，我们马上实施了这个方案，效果出奇得好：之前的受损分区全部恢复正常，业务数据得以正常生产和消费。
当然，给你分享这个案例的目的，并不是让你记住可以随意干掉/controller节点——这个操作其实是有一点危险的。事实上，我只是想通过这个真实的例子，向你说明，很多打开“精通Kafka之门”的钥匙是隐藏在源码中的。那么，接下来，我们就开始找“钥匙”吧。
## 集群元数据
想要完整地了解Controller的工作原理，我们首先就要学习它管理了哪些数据。毕竟，Controller的很多代码仅仅是做数据的管理操作而已。今天，我们就来重点学习Kafka集群元数据都有哪些。
如果说ZooKeeper是整个Kafka集群元数据的“真理之源（Source of Truth）”，那么Controller可以说是集群元数据的“真理之源副本（Backup Source of Truth）”。好吧，后面这个词是我自己发明的。你只需要理解，Controller承载了ZooKeeper上的所有元数据即可。
事实上，集群Broker是不会与ZooKeeper直接交互去获取元数据的。相反地，它们总是与Controller进行通信，获取和更新最新的集群数据。而且社区已经打算把ZooKeeper“干掉”了（我会在之后的“特别放送”里具体给你解释社区干掉ZooKeeper的操作），以后Controller将成为新的“真理之源”。
我们总说元数据，那么，到底什么是集群的元数据，或者说，Kafka集群的元数据都定义了哪些内容呢？我用一张图给你完整地展示一下，当前Kafka定义的所有集群元数据信息。
<img src="https://static001.geekbang.org/resource/image/f1/54/f146aceb78a5da31d887618303b5ff54.jpg" alt="">
可以看到，目前，Controller定义的元数据有17项之多。不过，并非所有的元数据都同等重要，你也不用完整地记住它们，我们只需要重点关注那些最重要的元数据，并结合源代码来了解下这些元数据都是用来做什么的。
在了解具体的元数据之前，我要先介绍下ControllerContext类。刚刚我们提到的这些元数据信息全部封装在这个类里。应该这么说，**这个类是Controller组件的数据容器类**。
## ControllerContext
Controller组件的源代码位于core包的src/main/scala/kafka/controller路径下，这里面有很多Scala源文件，**ControllerContext类就位于这个路径下的ControllerContext.scala文件中。**
该文件只有几百行代码，其中，最重要的数据结构就是ControllerContext类。前面说过，**它定义了前面提到的所有元数据信息，以及许多实用的工具方法**。比如，获取集群上所有主题分区对象的allPartitions方法、获取某主题分区副本列表的partitionReplicaAssignment方法，等等。
首先，我们来看下ControllerContext类的定义，如下所示：
```
class ControllerContext {
val stats = new ControllerStats // Controller统计信息类
var offlinePartitionCount = 0 // 离线分区计数器
val shuttingDownBrokerIds = mutable.Set.empty[Int] // 关闭中Broker的Id列表
private val liveBrokers = mutable.Set.empty[Broker] // 当前运行中Broker对象列表
private val liveBrokerEpochs = mutable.Map.empty[Int, Long] // 运行中Broker Epoch列表
var epoch: Int = KafkaController.InitialControllerEpoch // Controller当前Epoch值
var epochZkVersion: Int = KafkaController.InitialControllerEpochZkVersion // Controller对应ZooKeeper节点的Epoch值
val allTopics = mutable.Set.empty[String] // 集群主题列表
val partitionAssignments = mutable.Map.empty[String, mutable.Map[Int, ReplicaAssignment]] // 主题分区的副本列表
val partitionLeadershipInfo = mutable.Map.empty[TopicPartition, LeaderIsrAndControllerEpoch] // 主题分区的Leader/ISR副本信息
val partitionsBeingReassigned = mutable.Set.empty[TopicPartition] // 正处于副本重分配过程的主题分区列表
val partitionStates = mutable.Map.empty[TopicPartition, PartitionState] // 主题分区状态列表
val replicaStates = mutable.Map.empty[PartitionAndReplica, ReplicaState] // 主题分区的副本状态列表
val replicasOnOfflineDirs = mutable.Map.empty[Int, Set[TopicPartition]] // 不可用磁盘路径上的副本列表
val topicsToBeDeleted = mutable.Set.empty[String] // 待删除主题列表
val topicsWithDeletionStarted = mutable.Set.empty[String] // 已开启删除的主题列表
val topicsIneligibleForDeletion = mutable.Set.empty[String] // 暂时无法执行删除的主题列表
......
}
```
不多不少，这段代码中定义的字段正好17个，它们一一对应着上图中的那些元数据信息。下面，我选取一些重要的元数据，来详细解释下它们的含义。
这些元数据理解起来还是比较简单的，掌握了它们之后，你在理解MetadataCache，也就是元数据缓存的时候，就容易得多了。比如，接下来我要讲到的liveBrokers信息，就是Controller通过UpdateMetadataRequest请求同步给其他Broker的MetadataCache的。
### ControllerStats
第一个是ControllerStats类的变量。它的完整代码如下：
```
private[controller] class ControllerStats extends KafkaMetricsGroup {
// 统计每秒发生的Unclean Leader选举次数
val uncleanLeaderElectionRate = newMeter(&quot;UncleanLeaderElectionsPerSec&quot;, &quot;elections&quot;, TimeUnit.SECONDS)
// Controller事件通用的统计速率指标的方法
val rateAndTimeMetrics: Map[ControllerState, KafkaTimer] = ControllerState.values.flatMap { state =&gt;
state.rateAndTimeMetricName.map { metricName =&gt;
state -&gt; new KafkaTimer(newTimer(metricName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS))
}
}.toMap
}
```
顾名思义，它表征的是Controller的一些统计信息。目前，源码中定义了两大类统计指标：**UncleanLeaderElectionsPerSec和所有Controller事件状态的执行速率与时间**。
其中，**前者是计算Controller每秒执行的Unclean Leader选举数量，通常情况下，执行Unclean Leader选举可能造成数据丢失，一般不建议开启它**。一旦开启，你就需要时刻关注这个监控指标的值，确保Unclean Leader选举的速率维持在一个很低的水平，否则会出现很多数据丢失的情况。
**后者是统计所有Controller状态的速率和时间信息**，单位是毫秒。当前，Controller定义了很多事件，比如，TopicDeletion是执行主题删除的Controller事件、ControllerChange是执行Controller重选举的事件。ControllerStats的这个指标通过在每个事件名后拼接字符串RateAndTimeMs的方式，为每类Controller事件都创建了对应的速率监控指标。
由于Controller事件有很多种，对应的速率监控指标也有很多，有一些Controller事件是需要你额外关注的。
举个例子，IsrChangeNotification事件是标志ISR列表变更的事件，如果这个事件经常出现，说明副本的ISR列表经常发生变化，而这通常被认为是非正常情况，因此，你最好关注下这个事件的速率监控指标。
### offlinePartitionCount
**该字段统计集群中所有离线或处于不可用状态的主题分区数量**。所谓的不可用状态，就是我最开始举的例子中“Leader=-1”的情况。
ControllerContext中的updatePartitionStateMetrics方法根据**给定主题分区的当前状态和目标状态**，来判断该分区是否是离线状态的分区。如果是，则累加offlinePartitionCount字段的值，否则递减该值。方法代码如下：
```
// 更新offlinePartitionCount元数据
private def updatePartitionStateMetrics(
partition: TopicPartition,
currentState: PartitionState,
targetState: PartitionState): Unit = {
// 如果该主题当前并未处于删除中状态
if (!isTopicDeletionInProgress(partition.topic)) {
// targetState表示该分区要变更到的状态
// 如果当前状态不是OfflinePartition，即离线状态并且目标状态是离线状态
// 这个if语句判断是否要将该主题分区状态转换到离线状态
if (currentState != OfflinePartition &amp;&amp; targetState == OfflinePartition) {
offlinePartitionCount = offlinePartitionCount + 1
// 如果当前状态已经是离线状态，但targetState不是
// 这个else if语句判断是否要将该主题分区状态转换到非离线状态
} else if (currentState == OfflinePartition &amp;&amp; targetState != OfflinePartition) {
offlinePartitionCount = offlinePartitionCount - 1
}
}
}
```
该方法首先要判断，此分区所属的主题当前是否处于删除操作的过程中。如果是的话，Kafka就不能修改这个分区的状态，那么代码什么都不做，直接返回。否则，代码会判断该分区是否要转换到离线状态。如果targetState是OfflinePartition，那么就将offlinePartitionCount值加1，毕竟多了一个离线状态的分区。相反地，如果currentState是offlinePartition，而targetState反而不是，那么就将offlinePartitionCount值减1。
### shuttingDownBrokerIds
顾名思义，**该字段保存所有正在关闭中的Broker ID列表**。当Controller在管理集群Broker时，它要依靠这个字段来甄别Broker当前是否已关闭，因为处于关闭状态的Broker是不适合执行某些操作的，如分区重分配（Reassignment）以及主题删除等。
另外，Kafka必须要为这些关闭中的Broker执行很多清扫工作，Controller定义了一个onBrokerFailure方法，它就是用来做这个的。代码如下：
```
private def onBrokerFailure(deadBrokers: Seq[Int]): Unit = {
info(s&quot;Broker failure callback for ${deadBrokers.mkString(&quot;,&quot;)}&quot;)
// deadBrokers：给定的一组已终止运行的Broker Id列表
// 更新Controller元数据信息，将给定Broker从元数据的replicasOnOfflineDirs中移除
deadBrokers.foreach(controllerContext.replicasOnOfflineDirs.remove)
// 找出这些Broker上的所有副本对象
val deadBrokersThatWereShuttingDown =
deadBrokers.filter(id =&gt; controllerContext.shuttingDownBrokerIds.remove(id))
if (deadBrokersThatWereShuttingDown.nonEmpty)
info(s&quot;Removed ${deadBrokersThatWereShuttingDown.mkString(&quot;,&quot;)} from list of shutting down brokers.&quot;)
// 执行副本清扫工作
val allReplicasOnDeadBrokers = controllerContext.replicasOnBrokers(deadBrokers.toSet)
onReplicasBecomeOffline(allReplicasOnDeadBrokers)
// 取消这些Broker上注册的ZooKeeper监听器
unregisterBrokerModificationsHandler(deadBrokers)
}
```
该方法接收一组已终止运行的Broker ID列表，首先是更新Controller元数据信息，将给定Broker从元数据的replicasOnOfflineDirs和shuttingDownBrokerIds中移除，然后为这组Broker执行必要的副本清扫工作，也就是onReplicasBecomeOffline方法做的事情。
该方法主要依赖于分区状态机和副本状态机来完成对应的工作。在后面的课程中，我们会专门讨论副本状态机和分区状态机，这里你只要简单了解下它要做的事情就行了。后面等我们学完了这两个状态机之后，你可以再看下这个方法的具体实现原理。
这个方法的主要目的是把给定的副本标记成Offline状态，即不可用状态。具体分为以下这几个步骤：
1. 利用分区状态机将给定副本所在的分区标记为Offline状态；
1. 将集群上所有新分区和Offline分区状态变更为Online状态；
1. 将相应的副本对象状态变更为Offline。
### liveBrokers
**该字段保存当前所有运行中的Broker对象**。每个Broker对象就是一个&lt;Id，EndPoint，机架信息&gt;的三元组。ControllerContext中定义了很多方法来管理该字段，如addLiveBrokersAndEpochs、removeLiveBrokers和updateBrokerMetadata等。我拿updateBrokerMetadata方法进行说明，以下是源码：
```
def updateBrokerMetadata(oldMetadata: Broker, newMetadata: Broker): Unit = {
liveBrokers -= oldMetadata
liveBrokers += newMetadata
}
```
每当新增或移除已有Broker时，ZooKeeper就会更新其保存的Broker数据，从而引发Controller修改元数据，也就是会调用updateBrokerMetadata方法来增减Broker列表中的对象。怎么样，超简单吧？！
### liveBrokerEpochs
**该字段保存所有运行中Broker的Epoch信息**。Kafka使用Epoch数据防止Zombie Broker，即一个非常老的Broker被选举成为Controller。
另外，源码大多使用这个字段来获取所有运行中Broker的ID序号，如下面这个方法定义的那样：
```
def liveBrokerIds: Set[Int] = liveBrokerEpochs.keySet -- shuttingDownBrokerIds
```
liveBrokerEpochs的keySet方法返回Broker序号列表，然后从中移除关闭中的Broker序号，剩下的自然就是处于运行中的Broker序号列表了。
### epoch &amp; epochZkVersion
这两个字段一起说，因为它们都有“epoch”字眼，放在一起说，可以帮助你更好地理解两者的区别。epoch实际上就是ZooKeeper中/controller_epoch节点的值，你可以认为它就是Controller在整个Kafka集群的版本号，而epochZkVersion实际上是/controller_epoch节点的dataVersion值。
Kafka使用epochZkVersion来判断和防止Zombie Controller。这也就是说，原先在老Controller任期内的Controller操作在新Controller不能成功执行，因为新Controller的epochZkVersion要比老Controller的大。
另外，你可能会问：“这里的两个Epoch和上面的liveBrokerEpochs有啥区别呢？”实际上，这里的两个Epoch值都是属于Controller侧的数据，而liveBrokerEpochs是每个Broker自己的Epoch值。
### allTopics
**该字段保存集群上所有的主题名称**。每当有主题的增减，Controller就要更新该字段的值。
比如Controller有个processTopicChange方法，从名字上来看，它就是处理主题变更的。我们来看下它的代码实现，我把主要逻辑以注释的方式标注了出来：
```
private def processTopicChange(): Unit = {
if (!isActive) return // 如果Contorller已经关闭，直接返回
val topics = zkClient.getAllTopicsInCluster(true) // 从ZooKeeper中获取当前所有主题列表
val newTopics = topics -- controllerContext.allTopics // 找出当前元数据中不存在、ZooKeeper中存在的主题，视为新增主题
val deletedTopics = controllerContext.allTopics -- topics // 找出当前元数据中存在、ZooKeeper中不存在的主题，视为已删除主题
controllerContext.allTopics = topics // 更新Controller元数据
// 为新增主题和已删除主题执行后续处理操作
registerPartitionModificationsHandlers(newTopics.toSeq)
val addedPartitionReplicaAssignment = zkClient.getFullReplicaAssignmentForTopics(newTopics)
deletedTopics.foreach(controllerContext.removeTopic)
addedPartitionReplicaAssignment.foreach {
case (topicAndPartition, newReplicaAssignment) =&gt; controllerContext.updatePartitionFullReplicaAssignment(topicAndPartition, newReplicaAssignment)
}
info(s&quot;New topics: [$newTopics], deleted topics: [$deletedTopics], new partition replica assignment &quot; +
s&quot;[$addedPartitionReplicaAssignment]&quot;)
if (addedPartitionReplicaAssignment.nonEmpty)
onNewPartitionCreation(addedPartitionReplicaAssignment.keySet)
}
```
### partitionAssignments
**该字段保存所有主题分区的副本分配情况**。在我看来，**这是Controller最重要的元数据了**。事实上，你可以从这个字段衍生、定义很多实用的方法，来帮助Kafka从各种维度获取数据。
比如，如果Kafka要获取某个Broker上的所有分区，那么，它可以这样定义：
```
partitionAssignments.flatMap {
case (topic, topicReplicaAssignment) =&gt; topicReplicaAssignment.filter {
case (_, partitionAssignment) =&gt; partitionAssignment.replicas.contains(brokerId)
}.map {
case (partition, _) =&gt; new TopicPartition(topic, partition)
}
}.toSet
```
再比如，如果Kafka要获取某个主题的所有分区对象，代码可以这样写：
```
partitionAssignments.getOrElse(topic, mutable.Map.empty).map {
case (partition, _) =&gt; new TopicPartition(topic, partition)
}.toSet
```
实际上，这两段代码分别是ControllerContext.scala中partitionsOnBroker方法和partitionsForTopic两个方法的主体实现代码。
讲到这里，9个重要的元数据字段我就介绍完了。前面说过，ControllerContext中一共定义了17个元数据字段，你可以结合这9个字段，把其余8个的定义也过一遍，做到心中有数。**你对Controller元数据掌握得越好，就越能清晰地理解Controller在集群中发挥的作用**。
值得注意的是，在学习每个元数据字段时，除了它的定义之外，我建议你去搜索一下，与之相关的工具方法都是如何实现的。如果后面你想要新增获取或更新元数据的方法，你要对操作它们的代码有很强的把控力才行。
## 总结
今天，我们揭开了Kafka重要组件Controller的学习大幕。我给出了Controller模块的学习路线，还介绍了Controller的重要元数据。
- Controller元数据：Controller当前定义了17种元数据，涵盖Kafka集群数据的方方面面。
- ControllerContext：定义元数据以及操作它们的类。
- 关键元数据字段：最重要的元数据包括offlinePartitionCount、liveBrokers、partitionAssignments等。
- ControllerContext工具方法：ControllerContext 类定义了很多实用方法来管理这些元数据信息。
下节课，我们将学习Controller是如何给Broker发送请求的。Controller与Broker进行交互与通信，是Controller奠定王者地位的重要一环，我会向你详细解释它是如何做到这一点的。
## 课后讨论
我今天并未给出所有的元数据说明，请你自行结合代码分析一下，partitionLeadershipInfo里面保存的是什么数据？
欢迎你在留言区写下你的思考和答案，跟我交流讨论，也欢迎你把今天的内容分享给你的朋友。

385
极客时间专栏/Kafka核心源码解读/Controller模块/12 | ControllerChannelManager：Controller如何管理请求发送？.md Normal file
View File

@@ -0,0 +1,385 @@
<audio id="audio" title="12 | ControllerChannelManager：Controller如何管理请求发送？" controls="" preload="none"><source id="mp3" src="https://static001.geekbang.org/resource/audio/27/03/2788d6cb2b61d63a9dd239be95b61d03.mp3"></audio>
你好，我是胡夕。上节课，我们深入研究了ControllerContext.scala源码文件，掌握了Kafka集群定义的重要元数据。今天，我们来学习下Controller是如何给其他Broker发送请求的。
掌握了这部分实现原理，你就能更好地了解Controller究竟是如何与集群Broker进行交互，从而实现管理集群元数据的功能的。而且，阅读这部分源码，还能帮你定位和解决线上问题。我先跟你分享一个真实的案例。
当时还是在Kafka 0.10.0.1时代，我们突然发现，在线上环境中，很多元数据变更无法在集群的所有Broker上同步了。具体表现为，创建了主题后，有些Broker依然无法感知到。
我的第一感觉是Controller出现了问题，但又苦于无从排查和验证。后来，我想到，会不会是Controller端请求队列中积压的请求太多造成的呢？因为当时Controller所在的Broker本身承载着非常重的业务，这是非常有可能的原因。
在看了相关代码后，我们就在相应的源码中新加了一个监控指标，用于实时监控Controller的请求队列长度。当更新到生产环境后，我们很轻松地定位了问题。果然，由于Controller所在的Broker自身负载过大，导致Controller端的请求积压，从而造成了元数据更新的滞后。精准定位了问题之后，解决起来就很容易了。后来，社区于0.11版本正式引入了相关的监控指标。
你看，阅读源码，除了可以学习优秀开发人员编写的代码之外，我们还能根据自身的实际情况做定制化方案，实现一些非开箱即用的功能。
## Controller发送请求类型
下面，我们就正式进入到Controller请求发送管理部分的学习。你可能会问：“Controller也会给Broker发送请求吗？”当然！**Controller会给集群中的所有Broker（包括它自己所在的Broker）机器发送网络请求**。发送请求的目的，是让Broker执行相应的指令。我用一张图，来展示下Controller都会发送哪些请求，如下所示：
<img src="https://static001.geekbang.org/resource/image/3e/f7/3e8b0a34f003db5d67d5adafe8781ef7.jpg" alt="">
当前，Controller只会向Broker发送三类请求，分别是LeaderAndIsrRequest、StopReplicaRequest和UpdateMetadataRequest。注意，这里我使用的是“当前”！我只是说，目前仅有这三类，不代表以后不会有变化。事实上，我几乎可以肯定，以后能发送的RPC协议种类一定会变化的。因此，你需要掌握请求发送的原理。毕竟，所有请求发送都是通过相同的机制完成的。
还记得我在[第8节课](https://time.geekbang.org/column/article/232134)提到的控制类请求吗？没错，这三类请求就是典型的控制类请求。我来解释下它们的作用。
- LeaderAndIsrRequest：最主要的功能是，告诉Broker相关主题各个分区的Leader副本位于哪台Broker上、ISR中的副本都在哪些Broker上。在我看来，**它应该被赋予最高的优先级，毕竟，它有令数据类请求直接失效的本领**。试想一下，如果这个请求中的Leader副本变更了，之前发往老的Leader的PRODUCE请求是不是全部失效了？因此，我认为它是非常重要的控制类请求。
- StopReplicaRequest：告知指定Broker停止它上面的副本对象，该请求甚至还能删除副本底层的日志数据。这个请求主要的使用场景，是**分区副本迁移**和**删除主题**。在这两个场景下，都要涉及停掉Broker上的副本操作。
- UpdateMetadataRequest：顾名思义，该请求会更新Broker上的元数据缓存。集群上的所有元数据变更，都首先发生在Controller端，然后再经由这个请求广播给集群上的所有Broker。在我刚刚分享的案例中，正是因为这个请求被处理得不及时，才导致集群Broker无法获取到最新的元数据信息。
现在，社区越来越倾向于**将重要的数据结构源代码从服务器端的core工程移动到客户端的clients工程中**。这三类请求Java类的定义就封装在clients中，它们的抽象基类是AbstractControlRequest类，这个类定义了这三类请求的公共字段。
我用代码展示下这三类请求及其抽象父类的定义，以便让你对Controller发送的请求类型有个基本的认识。这些类位于clients工程下的src/main/java/org/apache/kafka/common/requests路径下。
先来看AbstractControlRequest类的主要代码：
```
public abstract class AbstractControlRequest extends AbstractRequest {
public static final long UNKNOWN_BROKER_EPOCH = -1L;
public static abstract class Builder&lt;T extends AbstractRequest&gt; extends AbstractRequest.Builder&lt;T&gt; {
protected final int controllerId;
protected final int controllerEpoch;
protected final long brokerEpoch;
......
}
```
区别于其他的数据类请求，抽象类请求必然包含3个字段。
- **controllerId**：Controller所在的Broker ID。
- **controllerEpoch**：Controller的版本信息。
- **brokerEpoch**：目标Broker的Epoch。
后面这两个Epoch字段用于隔离Zombie Controller和Zombie Broker，以保证集群的一致性。
在同一源码路径下，你能找到LeaderAndIsrRequest、StopReplicaRequest和UpdateMetadataRequest的定义，如下所示：
```
public class LeaderAndIsrRequest extends AbstractControlRequest { ...... }
public class StopReplicaRequest extends AbstractControlRequest { ...... }
public class UpdateMetadataRequest extends AbstractControlRequest { ...... }
```
## RequestSendThread
说完了Controller发送什么请求，接下来我们说说怎么发。
Kafka源码非常喜欢生产者-消费者模式。该模式的好处在于，**解耦生产者和消费者逻辑，分离两者的集中性交互**。学完了“请求处理”模块，现在，你一定很赞同这个说法吧。还记得Broker端的SocketServer组件吗？它就在内部定义了一个线程共享的请求队列：它下面的Processor线程扮演Producer，而KafkaRequestHandler线程扮演Consumer。
对于Controller而言，源码同样使用了这个模式：它依然是一个线程安全的阻塞队列，Controller事件处理线程（第13节课会详细说它）负责向这个队列写入待发送的请求，而一个名为RequestSendThread的线程负责执行真正的请求发送。如下图所示：
<img src="https://static001.geekbang.org/resource/image/82/21/825d084eb1517daace5532d1c93b0321.jpg" alt="">
Controller会为集群中的每个Broker都创建一个对应的RequestSendThread线程。Broker上的这个线程，持续地从阻塞队列中获取待发送的请求。
那么，Controller往阻塞队列上放什么数据呢？这其实是由源码中的QueueItem类定义的。代码如下：
```
case class QueueItem(apiKey: ApiKeys, request: AbstractControlRequest.Builder[_ &lt;: AbstractControlRequest], callback: AbstractResponse =&gt; Unit, enqueueTimeMs: Long)
```
每个QueueItem的核心字段都是**AbstractControlRequest.Builder对象**。你基本上可以认为，它就是阻塞队列上AbstractControlRequest类型。
需要注意的是这里的“&lt;:”符号，它在Scala中表示**上边界**的意思，即字段request必须是AbstractControlRequest的子类，也就是上面说到的那三类请求。
这也就是说，每个QueueItem实际保存的都是那三类请求中的其中一类。如果使用一个BlockingQueue对象来保存这些QueueItem，那么，代码就实现了一个请求阻塞队列。这就是RequestSendThread类做的事情。
接下来，我们就来学习下RequestSendThread类的定义。我给一些主要的字段添加了注释。
```
class RequestSendThread(val controllerId: Int, // Controller所在Broker的Id
val controllerContext: ControllerContext, // Controller元数据信息
val queue: BlockingQueue[QueueItem], // 请求阻塞队列
val networkClient: NetworkClient, // 用于执行发送的网络I/O类
val brokerNode: Node, // 目标Broker节点
val config: KafkaConfig, // Kafka配置信息
val time: Time,
val requestRateAndQueueTimeMetrics: Timer,
val stateChangeLogger: StateChangeLogger,
name: String) extends ShutdownableThread(name = name) {
......
}
```
其实，RequestSendThread最重要的是它的**doWork方法**，也就是执行线程逻辑的方法：
```
override def doWork(): Unit = {
def backoff(): Unit = pause(100, TimeUnit.MILLISECONDS)
val QueueItem(apiKey, requestBuilder, callback, enqueueTimeMs) = queue.take() // 以阻塞的方式从阻塞队列中取出请求
requestRateAndQueueTimeMetrics.update(time.milliseconds() - enqueueTimeMs, TimeUnit.MILLISECONDS) // 更新统计信息
var clientResponse: ClientResponse = null
try {
var isSendSuccessful = false
while (isRunning &amp;&amp; !isSendSuccessful) {
try {
// 如果没有创建与目标Broker的TCP连接，或连接暂时不可用
if (!brokerReady()) {
isSendSuccessful = false
backoff() // 等待重试
}
else {
val clientRequest = networkClient.newClientRequest(brokerNode.idString, requestBuilder,
time.milliseconds(), true)
// 发送请求，等待接收Response
clientResponse = NetworkClientUtils.sendAndReceive(networkClient, clientRequest, time)
isSendSuccessful = true
}
} catch {
case e: Throwable =&gt;
warn(s&quot;Controller $controllerId epoch ${controllerContext.epoch} fails to send request $requestBuilder &quot; +
s&quot;to broker $brokerNode. Reconnecting to broker.&quot;, e)
// 如果出现异常，关闭与对应Broker的连接
networkClient.close(brokerNode.idString)
isSendSuccessful = false
backoff()
}
}
// 如果接收到了Response
if (clientResponse != null) {
val requestHeader = clientResponse.requestHeader
val api = requestHeader.apiKey
// 此Response的请求类型必须是LeaderAndIsrRequest、StopReplicaRequest或UpdateMetadataRequest中的一种
if (api != ApiKeys.LEADER_AND_ISR &amp;&amp; api != ApiKeys.STOP_REPLICA &amp;&amp; api != ApiKeys.UPDATE_METADATA)
throw new KafkaException(s&quot;Unexpected apiKey received: $apiKey&quot;)
val response = clientResponse.responseBody
stateChangeLogger.withControllerEpoch(controllerContext.epoch)
.trace(s&quot;Received response &quot; +
s&quot;${response.toString(requestHeader.apiVersion)} for request $api with correlation id &quot; +
s&quot;${requestHeader.correlationId} sent to broker $brokerNode&quot;)
if (callback != null) {
callback(response) // 处理回调
}
}
} catch {
case e: Throwable =&gt;
error(s&quot;Controller $controllerId fails to send a request to broker $brokerNode&quot;, e)
networkClient.close(brokerNode.idString)
}
}
```
我用一张图来说明doWork的执行逻辑：
<img src="https://static001.geekbang.org/resource/image/86/19/869727e22f882509a149d1065a8a1719.jpg" alt="">
总体上来看，doWork的逻辑很直观。它的主要作用是从阻塞队列中取出待发送的请求，然后把它发送出去，之后等待Response的返回。在等待Response的过程中，线程将一直处于阻塞状态。当接收到Response之后，调用callback执行请求处理完成后的回调逻辑。
需要注意的是，RequestSendThread线程对请求发送的处理方式与Broker处理请求不太一样。它调用的sendAndReceive方法在发送完请求之后，会原地进入阻塞状态，等待Response返回。只有接收到Response，并执行完回调逻辑之后，该线程才能从阻塞队列中取出下一个待发送请求进行处理。
## ControllerChannelManager
了解了RequestSendThread线程的源码之后，我们进入到ControllerChannelManager类的学习。
这个类和RequestSendThread是合作共赢的关系。在我看来，它有两大类任务。
- 管理Controller与集群Broker之间的连接，并为每个Broker创建RequestSendThread线程实例；
- 将要发送的请求放入到指定Broker的阻塞队列中，等待该Broker专属的RequestSendThread线程进行处理。
由此可见，它们是紧密相连的。
ControllerChannelManager类最重要的数据结构是brokerStateInfo，它是在下面这行代码中定义的：
```
protected val brokerStateInfo = new HashMap[Int, ControllerBrokerStateInfo]
```
这是一个HashMap类型，Key是Integer类型，其实就是集群中Broker的ID信息，而Value是一个ControllerBrokerStateInfo。
你可能不太清楚ControllerBrokerStateInfo类是什么，我先解释一下。它本质上是一个POJO类，仅仅是承载若干数据结构的容器，如下所示：
```
case class ControllerBrokerStateInfo(networkClient: NetworkClient,
brokerNode: Node,
messageQueue: BlockingQueue[QueueItem],
requestSendThread: RequestSendThread,
queueSizeGauge: Gauge[Int],
requestRateAndTimeMetrics: Timer,
reconfigurableChannelBuilder: Option[Reconfigurable])
```
它有三个非常关键的字段。
- **brokerNode**：目标Broker节点对象，里面封装了目标Broker的连接信息，比如主机名、端口号等。
- **messageQueue**：请求消息阻塞队列。你可以发现，Controller为每个目标Broker都创建了一个消息队列。
- **requestSendThread**：Controller使用这个线程给目标Broker发送请求。
你一定要记住这三个字段，因为它们是实现Controller发送请求的关键因素。
为什么呢？我们思考一下，如果Controller要给Broker发送请求，肯定需要解决三个问题：发给谁？发什么？怎么发？“发给谁”就是由brokerNode决定的；messageQueue里面保存了要发送的请求，因而解决了“发什么”的问题；最后的“怎么发”就是依赖requestSendThread变量实现的。
好了，我们现在回到ControllerChannelManager。它定义了5个public方法，我来一一介绍下。
- **startup方法**：Controller组件在启动时，会调用ControllerChannelManager的startup方法。该方法会从元数据信息中找到集群的Broker列表，然后依次为它们调用addBroker方法，把它们加到brokerStateInfo变量中，最后再依次启动brokerStateInfo中的RequestSendThread线程。
- **shutdown方法**：关闭所有RequestSendThread线程，并清空必要的资源。
- **sendRequest方法**：从名字看，就是发送请求，实际上就是把请求对象提交到请求队列。
- **addBroker方法**：添加目标Broker到brokerStateInfo数据结构中，并创建必要的配套资源，如请求队列、RequestSendThread线程对象等。最后，RequestSendThread启动线程。
- **removeBroker方法**：从brokerStateInfo移除目标Broker的相关数据。
这里面大部分的方法逻辑都很简单，从方法名字就可以看得出来。我重点说一下**addBroker**，以及**底层相关的私有方法addNewBroker和startRequestSendThread方法**。
毕竟，addBroker是最重要的逻辑。每当集群中扩容了新的Broker时，Controller就会调用这个方法为新Broker增加新的RequestSendThread线程。
我们先来看addBroker：
```
def addBroker(broker: Broker): Unit = {
brokerLock synchronized {
// 如果该Broker是新Broker的话
if (!brokerStateInfo.contains(broker.id)) {
// 将新Broker加入到Controller管理，并创建对应的RequestSendThread线程
addNewBroker(broker)
// 启动RequestSendThread线程
startRequestSendThread(broker.id)
}
}
}
```
整个代码段被brokerLock保护起来了。还记得brokerStateInfo的定义吗？它仅仅是一个HashMap对象，因为不是线程安全的，所以任何访问该变量的地方，都需要锁的保护。
这段代码的逻辑是，判断目标Broker的序号，是否已经保存在brokerStateInfo中。如果是，就说明这个Broker之前已经添加过了，就没必要再次添加了；否则，addBroker方法会对目前的Broker执行两个操作：
1. 把该Broker节点添加到brokerStateInfo中；
1. 启动与该Broker对应的RequestSendThread线程。
这两步分别是由addNewBroker和startRequestSendThread方法实现的。
addNewBroker方法的逻辑比较复杂，我用注释的方式给出主要步骤：
```
private def addNewBroker(broker: Broker): Unit = {
// 为该Broker构造请求阻塞队列
val messageQueue = new LinkedBlockingQueue[QueueItem]
debug(s&quot;Controller ${config.brokerId} trying to connect to broker ${broker.id}&quot;)
val controllerToBrokerListenerName = config.controlPlaneListenerName.getOrElse(config.interBrokerListenerName)
val controllerToBrokerSecurityProtocol = config.controlPlaneSecurityProtocol.getOrElse(config.interBrokerSecurityProtocol)
// 获取待连接Broker节点对象信息
val brokerNode = broker.node(controllerToBrokerListenerName)
val logContext = new LogContext(s&quot;[Controller id=${config.brokerId}, targetBrokerId=${brokerNode.idString}] &quot;)
val (networkClient, reconfigurableChannelBuilder) = {
val channelBuilder = ChannelBuilders.clientChannelBuilder(
controllerToBrokerSecurityProtocol,
JaasContext.Type.SERVER,
config,
controllerToBrokerListenerName,
config.saslMechanismInterBrokerProtocol,
time,
config.saslInterBrokerHandshakeRequestEnable,
logContext
)
val reconfigurableChannelBuilder = channelBuilder match {
case reconfigurable: Reconfigurable =&gt;
config.addReconfigurable(reconfigurable)
Some(reconfigurable)
case _ =&gt; None
}
// 创建NIO Selector实例用于网络数据传输
val selector = new Selector(
NetworkReceive.UNLIMITED,
Selector.NO_IDLE_TIMEOUT_MS,
metrics,
time,
&quot;controller-channel&quot;,
Map(&quot;broker-id&quot; -&gt; brokerNode.idString).asJava,
false,
channelBuilder,
logContext
)
// 创建NetworkClient实例
// NetworkClient类是Kafka clients工程封装的顶层网络客户端API
// 提供了丰富的方法实现网络层IO数据传输
val networkClient = new NetworkClient(
selector,
new ManualMetadataUpdater(Seq(brokerNode).asJava),
config.brokerId.toString,
1,
0,
0,
Selectable.USE_DEFAULT_BUFFER_SIZE,
Selectable.USE_DEFAULT_BUFFER_SIZE,
config.requestTimeoutMs,
ClientDnsLookup.DEFAULT,
time,
false,
new ApiVersions,
logContext
)
(networkClient, reconfigurableChannelBuilder)
}
// 为这个RequestSendThread线程设置线程名称
val threadName = threadNamePrefix match {
case None =&gt; s&quot;Controller-${config.brokerId}-to-broker-${broker.id}-send-thread&quot;
case Some(name) =&gt; s&quot;$name:Controller-${config.brokerId}-to-broker-${broker.id}-send-thread&quot;
}
// 构造请求处理速率监控指标
val requestRateAndQueueTimeMetrics = newTimer(
RequestRateAndQueueTimeMetricName, TimeUnit.MILLISECONDS, TimeUnit.SECONDS, brokerMetricTags(broker.id)
)
// 创建RequestSendThread实例
val requestThread = new RequestSendThread(config.brokerId, controllerContext, messageQueue, networkClient,
brokerNode, config, time, requestRateAndQueueTimeMetrics, stateChangeLogger, threadName)
requestThread.setDaemon(false)
val queueSizeGauge = newGauge(QueueSizeMetricName, () =&gt; messageQueue.size, brokerMetricTags(broker.id))
// 创建该Broker专属的ControllerBrokerStateInfo实例
// 并将其加入到brokerStateInfo统一管理
brokerStateInfo.put(broker.id, ControllerBrokerStateInfo(networkClient, brokerNode, messageQueue,
requestThread, queueSizeGauge, requestRateAndQueueTimeMetrics, reconfigurableChannelBuilder))
}
```
为了方便你理解，我还画了一张流程图形象说明它的执行流程：
<img src="https://static001.geekbang.org/resource/image/4f/22/4f34c319f9480c16ac12dee78d5ba322.jpg" alt="">
addNewBroker的关键在于，**要为目标Broker创建一系列的配套资源**，比如，NetworkClient用于网络I/O操作、messageQueue用于阻塞队列、requestThread用于发送请求，等等。
至于startRequestSendThread方法，就简单得多了，只有几行代码而已。
```
protected def startRequestSendThread(brokerId: Int): Unit = {
// 获取指定Broker的专属RequestSendThread实例
val requestThread = brokerStateInfo(brokerId).requestSendThread
if (requestThread.getState == Thread.State.NEW)
// 启动线程
requestThread.start()
}
```
它首先根据给定的Broker序号信息，从brokerStateInfo中找出对应的ControllerBrokerStateInfo对象。有了这个对象，也就有了为该目标Broker服务的所有配套资源。下一步就是从ControllerBrokerStateInfo中拿出RequestSendThread对象，再启动它就好了。
## 总结
今天，我结合ControllerChannelManager.scala文件，重点分析了Controller向Broker发送请求机制的实现原理。
Controller主要通过ControllerChannelManager类来实现与其他Broker之间的请求发送。其中，ControllerChannelManager类中定义的RequestSendThread是主要的线程实现类，用于实际发送请求给集群Broker。除了RequestSendThread之外，ControllerChannelManager还定义了相应的管理方法，如添加Broker、移除Broker等。通过这些管理方法，Controller在集群扩缩容时能够快速地响应到这些变化，完成对应Broker连接的创建与销毁。
我们来回顾下这节课的重点。
- Controller端请求：Controller发送三类请求给Broker，分别是LeaderAndIsrRequest、StopReplicaRequest和UpdateMetadataRequest。
- RequestSendThread：该线程负责将请求发送给集群中的相关或所有Broker。
- 请求阻塞队列+RequestSendThread：Controller会为集群上所有Broker创建对应的请求阻塞队列和RequestSendThread线程。
其实，今天讲的所有东西都只是这节课的第二张图中“消费者”的部分，我们并没有详细了解请求是怎么被放到请求队列中的。接下来，我们就会针对这个问题，深入地去探讨Controller单线程的事件处理器是如何实现的。
<img src="https://static001.geekbang.org/resource/image/00/b9/00fce28d26a94389f2bb5e957b650bb9.jpg" alt="">
## 课后讨论
你觉得，为每个Broker都创建一个RequestSendThread的方案有什么优缺点？
欢迎你在留言区写下你的思考和答案，跟我交流讨论，也欢迎你把今天的内容分享给你的朋友。

350
极客时间专栏/Kafka核心源码解读/Controller模块/13 | ControllerEventManager：变身单线程后的Controller如何处理事件？.md Normal file
View File

@@ -0,0 +1,350 @@
<audio id="audio" title="13 | ControllerEventManager：变身单线程后的Controller如何处理事件？" controls="" preload="none"><source id="mp3" src="https://static001.geekbang.org/resource/audio/35/15/3578ea76868ab16b2eab2a4512646b15.mp3"></audio>
你好，我是胡夕。 今天，我们来学习下Controller的单线程事件处理器源码。
所谓的单线程事件处理器，就是Controller端定义的一个组件。该组件内置了一个专属线程，负责处理其他线程发送过来的Controller事件。另外，它还定义了一些管理方法，用于为专属线程输送待处理事件。
在0.11.0.0版本之前，Controller组件的源码非常复杂。集群元数据信息在程序中同时被多个线程访问，因此，源码里有大量的Monitor锁、Lock锁或其他线程安全机制，这就导致，这部分代码读起来晦涩难懂，改动起来也困难重重，因为你根本不知道，变动了这个线程访问的数据，会不会影响到其他线程。同时，开发人员在修复Controller Bug时，也非常吃力。
鉴于这个原因，自0.11.0.0版本开始，社区陆续对Controller代码结构进行了改造。其中非常重要的一环，就是将**多线程并发访问的方式改为了单线程的事件队列方式**。
这里的单线程，并非是指Controller只有一个线程了，而是指**对局部状态的访问限制在一个专属线程上**，即让这个特定线程排他性地操作Controller元数据信息。
这样一来，整个组件代码就不必担心多线程访问引发的各种线程安全问题了，源码也可以抛弃各种不必要的锁机制，最终大大简化了Controller端的代码结构。
这部分源码非常重要，**它能够帮助你掌握Controller端处理各类事件的原理，这将极大地提升你在实际场景中处理Controller各类问题的能力**。因此，我建议你多读几遍，彻底了解Controller是怎么处理各种事件的。
## 基本术语和概念
接下来，我们先宏观领略一下Controller单线程事件队列处理模型及其基础组件。
<img src="https://static001.geekbang.org/resource/image/67/31/67fbf8a12ebb57bc309188dcbc18e231.jpg" alt="">
从图中可见，Controller端有多个线程向事件队列写入不同种类的事件，比如，ZooKeeper端注册的Watcher线程、KafkaRequestHandler线程、Kafka定时任务线程，等等。而在事件队列的另一端，只有一个名为ControllerEventThread的线程专门负责“消费”或处理队列中的事件。这就是所谓的单线程事件队列模型。
参与实现这个模型的源码类有4个。
- ControllerEventProcessor：Controller端的事件处理器接口。
- ControllerEvent：Controller事件，也就是事件队列中被处理的对象。
- ControllerEventManager：事件处理器，用于创建和管理ControllerEventThread。
- ControllerEventThread：专属的事件处理线程，唯一的作用是处理不同种类的ControllEvent。这个类是ControllerEventManager类内部定义的线程类。
今天，我们的重要目标就是要搞懂这4个类。就像我前面说的，它们完整地构建出了单线程事件队列模型。下面我们将一个一个地学习它们的源码，你要重点掌握事件队列的实现以及专属线程是如何访问事件队列的。
## ControllerEventProcessor
这个接口位于controller包下的ControllerEventManager.scala文件中。它定义了一个支持普通处理和抢占处理Controller事件的接口，代码如下所示：
```
trait ControllerEventProcessor {
def process(event: ControllerEvent): Unit
def preempt(event: ControllerEvent): Unit
}
```
该接口定义了两个方法，分别是process和preempt。
- process：接收一个Controller事件，并进行处理。
- preempt：接收一个Controller事件，并抢占队列之前的事件进行优先处理。
目前，在Kafka源码中，KafkaController类是Controller组件的功能实现类，它也是ControllerEventProcessor接口的唯一实现类。
对于这个接口，你要重点掌握process方法的作用，因为**它是实现Controller事件处理的主力方法**。你要了解process方法**处理各类Controller事件的代码结构是什么样的**，而且还要能够准确地找到处理每类事件的子方法。
至于preempt方法，你仅需要了解，Kafka使用它实现某些高优先级事件的抢占处理即可，毕竟，目前在源码中只有两类事件（ShutdownEventThread和Expire）需要抢占式处理，出镜率不是很高。
## ControllerEvent
这就是前面说到的Controller事件，在源码中对应的就是ControllerEvent接口。该接口定义在KafkaController.scala文件中，本质上是一个trait类型，如下所示：
```
sealed trait ControllerEvent {
def state: ControllerState
}
```
**每个ControllerEvent都定义了一个状态**。Controller在处理具体的事件时，会对状态进行相应的变更。这个状态是由源码文件ControllerState.scala中的抽象类ControllerState定义的，代码如下：
```
sealed abstract class ControllerState {
def value: Byte
def rateAndTimeMetricName: Option[String] =
if (hasRateAndTimeMetric) Some(s&quot;${toString}RateAndTimeMs&quot;) else None
protected def hasRateAndTimeMetric: Boolean = true
}
```
每类ControllerState都定义一个value值，表示Controller状态的序号，从0开始。另外，rateAndTimeMetricName方法是用于构造Controller状态速率的监控指标名称的。
比如，TopicChange是一类ControllerState，用于表示主题总数发生了变化。为了监控这类状态变更速率，代码中的rateAndTimeMetricName方法会定义一个名为TopicChangeRateAndTimeMs的指标。当然，并非所有的ControllerState都有对应的速率监控指标，比如，表示空闲状态的Idle就没有对应的指标。
目前，Controller总共定义了25类事件和17种状态，它们的对应关系如下表所示：
<img src="https://static001.geekbang.org/resource/image/a4/63/a4bd821a8fac58bdf9c813379bc28e63.jpg" alt="">
内容看着好像有很多，那我们应该怎样使用这张表格呢？
实际上，你并不需要记住每一行的对应关系。这张表格更像是一个工具，当你监控到某些Controller状态变更速率异常的时候，你可以通过这张表格，快速确定可能造成瓶颈的Controller事件，并定位处理该事件的函数代码，辅助你进一步地调试问题。
另外，你要了解的是，**多个ControllerEvent可能归属于相同的ControllerState。**
比如，TopicChange和PartitionModifications事件都属于TopicChange状态，毕竟，它们都与Topic的变更有关。前者是创建Topic，后者是修改Topic的属性，比如，分区数或副本因子，等等。
再比如，BrokerChange和BrokerModifications事件都属于 BrokerChange状态，表征的都是对Broker属性的修改。
## ControllerEventManager
有了这些铺垫，我们就可以开始学习事件处理器的实现代码了。
在Kafka中，Controller事件处理器代码位于controller包下的ControllerEventManager.scala文件下。我用一张图来展示下这个文件的结构：
<img src="https://static001.geekbang.org/resource/image/11/4b/1137cfd21025c797369fa3d39cee5d4b.jpg" alt="">
如图所示，该文件主要由4个部分组成。
- **ControllerEventManager Object**：保存一些字符串常量，比如线程名字。
- **ControllerEventProcessor**：前面讲过的事件处理器接口，目前只有KafkaController实现了这个接口。
- **QueuedEvent**：表征事件队列上的事件对象。
- **ControllerEventManager Class**：ControllerEventManager的伴生类，主要用于创建和管理事件处理线程和事件队列。就像我前面说的，这个类中定义了重要的ControllerEventThread线程类，还有一些其他值得我们学习的重要方法，一会儿我们详细说说。
ControllerEventManager对象仅仅定义了3个公共变量，没有任何逻辑，你简单看下就行。至于ControllerEventProcessor接口，我们刚刚已经学习过了。接下来，我们重点学习后面这两个类。
### QueuedEvent
我们先来看QueuedEvent的定义，全部代码如下：
```
// 每个QueuedEvent定义了两个字段
// event: ControllerEvent类，表示Controller事件
// enqueueTimeMs：表示Controller事件被放入到事件队列的时间戳
class QueuedEvent(val event: ControllerEvent,
val enqueueTimeMs: Long) {
// 标识事件是否开始被处理
val processingStarted = new CountDownLatch(1)
// 标识事件是否被处理过
val spent = new AtomicBoolean(false)
// 处理事件
def process(processor: ControllerEventProcessor): Unit = {
if (spent.getAndSet(true))
return
processingStarted.countDown()
processor.process(event)
}
// 抢占式处理事件
def preempt(processor: ControllerEventProcessor): Unit = {
if (spent.getAndSet(true))
return
processor.preempt(event)
}
// 阻塞等待事件被处理完成
def awaitProcessing(): Unit = {
processingStarted.await()
}
override def toString: String = {
s&quot;QueuedEvent(event=$event, enqueueTimeMs=$enqueueTimeMs)&quot;
}
}
```
可以看到，每个QueuedEvent对象实例都裹挟了一个ControllerEvent。另外，每个QueuedEvent还定义了process、preempt和awaitProcessing方法，分别表示**处理事件**、**以抢占方式处理事件**，以及**等待事件处理**。
其中，process方法和preempt方法的实现原理，就是调用给定ControllerEventProcessor接口的process和preempt方法，非常简单。
在QueuedEvent对象中，我们再一次看到了CountDownLatch的身影，我在[第7节课](https://time.geekbang.org/column/article/231139)里提到过它。Kafka源码非常喜欢用CountDownLatch来做各种条件控制，比如用于侦测线程是否成功启动、成功关闭，等等。
在这里，QueuedEvent使用它的唯一目的，是确保Expire事件在建立ZooKeeper会话前被处理。
如果不是在这个场景下，那么，代码就用spent来标识该事件是否已经被处理过了，如果已经被处理过了，再次调用process方法时就会直接返回，什么都不做。
### ControllerEventThread
了解了QueuedEvent，我们来看下消费它们的ControllerEventThread类。
首先是这个类的定义代码：
```
class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptible = false) {
logIdent = s&quot;[ControllerEventThread controllerId=$controllerId] &quot;
......
}
```
这个类就是一个普通的线程类，继承了ShutdownableThread基类，而后者是Kafka为很多线程类定义的公共父类。该父类是Java Thread类的子类，其线程逻辑方法run的主要代码如下：
```
def doWork(): Unit
override def run(): Unit = {
......
try {
while (isRunning)
doWork()
} catch {
......
}
......
}
```
可见，这个父类会循环地执行doWork方法的逻辑，而该方法的具体实现则交由子类来完成。
作为Controller唯一的事件处理线程，我们要时刻关注这个线程的运行状态。因此，我们必须要知道这个线程在JVM上的名字，这样后续我们就能有针对性地对其展开监控。这个线程的名字是由ControllerEventManager Object中ControllerEventThreadName变量定义的，如下所示：
```
object ControllerEventManager {
val ControllerEventThreadName = &quot;controller-event-thread&quot;
......
}
```
现在我们看看ControllerEventThread类的doWork是如何实现的。代码如下：
```
override def doWork(): Unit = {
// 从事件队列中获取待处理的Controller事件，否则等待
val dequeued = queue.take()
dequeued.event match {
// 如果是关闭线程事件，什么都不用做。关闭线程由外部来执行
case ShutdownEventThread =&gt;
case controllerEvent =&gt;
_state = controllerEvent.state
// 更新对应事件在队列中保存的时间
eventQueueTimeHist.update(time.milliseconds() - dequeued.enqueueTimeMs)
try {
def process(): Unit = dequeued.process(processor)
// 处理事件，同时计算处理速率
rateAndTimeMetrics.get(state) match {
case Some(timer) =&gt; timer.time { process() }
case None =&gt; process()
}
} catch {
case e: Throwable =&gt; error(s&quot;Uncaught error processing event $controllerEvent&quot;, e)
}
_state = ControllerState.Idle
}
}
```
我用一张图来展示下具体的执行流程：
<img src="https://static001.geekbang.org/resource/image/db/d1/db4905db1a32ac7d356317f29d920dd1.jpg" alt="">
大体上看，执行逻辑很简单。
首先是调用LinkedBlockingQueue的take方法，去获取待处理的QueuedEvent对象实例。注意，这里用的是**take方法**，这说明，如果事件队列中没有QueuedEvent，那么，ControllerEventThread线程将一直处于阻塞状态，直到事件队列上插入了新的待处理事件。
一旦拿到QueuedEvent事件后，线程会判断是否是ShutdownEventThread事件。当ControllerEventManager关闭时，会显式地向事件队列中塞入ShutdownEventThread，表明要关闭ControllerEventThread线程。如果是该事件，那么ControllerEventThread什么都不用做，毕竟要关闭这个线程了。相反地，如果是其他的事件，就调用QueuedEvent的process方法执行对应的处理逻辑，同时计算事件被处理的速率。
该process方法底层调用的是ControllerEventProcessor的process方法，如下所示：
```
def process(processor: ControllerEventProcessor): Unit = {
// 若已经被处理过，直接返回
if (spent.getAndSet(true))
return
processingStarted.countDown()
// 调用ControllerEventProcessor的process方法处理事件
processor.process(event)
}
```
方法首先会判断该事件是否已经被处理过，如果是，就直接返回；如果不是，就调用ControllerEventProcessor的process方法处理事件。
你可能很关心，每个ControllerEventProcessor的process方法是在哪里实现的？实际上，它们都封装在KafkaController.scala文件中。还记得我之前说过，KafkaController类是目前源码中ControllerEventProcessor接口的唯一实现类吗？
实际上，就是KafkaController类实现了ControllerEventProcessor的process方法。由于代码过长，而且有很多重复结构的代码，因此，我只展示部分代码：
```
override def process(event: ControllerEvent): Unit = {
try {
// 依次匹配ControllerEvent事件
event match {
case event: MockEvent =&gt;
event.process()
case ShutdownEventThread =&gt;
error(&quot;Received a ShutdownEventThread event. This type of event is supposed to be handle by ControllerEventThread&quot;)
case AutoPreferredReplicaLeaderElection =&gt;
processAutoPreferredReplicaLeaderElection()
......
}
} catch {
// 如果Controller换成了别的Broker
case e: ControllerMovedException =&gt;
info(s&quot;Controller moved to another broker when processing $event.&quot;, e)
// 执行Controller卸任逻辑
maybeResign()
case e: Throwable =&gt;
error(s&quot;Error processing event $event&quot;, e)
} finally {
updateMetrics()
}
}
```
这个process方法接收一个ControllerEvent实例，接着会判断它是哪类Controller事件，并调用相应的处理方法。比如，如果是AutoPreferredReplicaLeaderElection事件，则调用processAutoPreferredReplicaLeaderElection方法；如果是其他类型的事件，则调用process***方法。
### 其他方法
除了QueuedEvent和ControllerEventThread之外，**put方法**和**clearAndPut方法也很重要**。如果说ControllerEventThread是读取队列事件的，那么，这两个方法就是向队列生产元素的。
在这两个方法中，put是把指定ControllerEvent插入到事件队列，而clearAndPut则是先执行具有高优先级的抢占式事件，之后清空队列所有事件，最后再插入指定的事件。
下面这两段源码分别对应于这两个方法：
```
// put方法
def put(event: ControllerEvent): QueuedEvent = inLock(putLock) {
// 构建QueuedEvent实例
val queuedEvent = new QueuedEvent(event, time.milliseconds())
// 插入到事件队列
queue.put(queuedEvent)
// 返回新建QueuedEvent实例
queuedEvent
}
// clearAndPut方法
def clearAndPut(event: ControllerEvent): QueuedEvent = inLock(putLock) {
// 优先处理抢占式事件
queue.forEach(_.preempt(processor))
// 清空事件队列
queue.clear()
// 调用上面的put方法将给定事件插入到事件队列
put(event)
}
```
整体上代码很简单，需要解释的地方不多，但我想和你讨论一个问题。
你注意到，源码中的put方法使用putLock对代码进行保护了吗？
就我个人而言，我觉得这个putLock是不需要的，因为LinkedBlockingQueue数据结构本身就已经是线程安全的了。put方法只会与全局共享变量queue打交道，因此，它们的线程安全性完全可以委托LinkedBlockingQueue实现。更何况，LinkedBlockingQueue内部已经维护了一个putLock和一个takeLock，专门保护读写操作。
当然，我同意在clearAndPut中使用锁的做法，毕竟，我们要保证，访问抢占式事件和清空操作构成一个原子操作。
## 总结
今天，我们重点学习了Controller端的单线程事件队列实现方式，即ControllerEventManager通过构建ControllerEvent、ControllerState和对应的ControllerEventThread线程，并且结合专属事件队列，共同实现事件处理。我们来回顾下这节课的重点。
- ControllerEvent：定义Controller能够处理的各类事件名称，目前总共定义了25类事件。
- ControllerState：定义Controller状态。你可以认为，它是ControllerEvent的上一级分类，因此，ControllerEvent和ControllerState是多对一的关系。
- ControllerEventManager：Controller定义的事件管理器，专门定义和维护专属线程以及对应的事件队列。
- ControllerEventThread：事件管理器创建的事件处理线程。该线程排他性地读取事件队列并处理队列中的所有事件。
<img src="https://static001.geekbang.org/resource/image/4e/26/4ec79e1ff2b83d0a1e850b6acf30b226.jpg" alt="">
下节课，我们将正式进入到KafkaController的学习。这是一个有着2100多行的大文件，不过大部分的代码都是实现那27类ControllerEvent的处理逻辑，因此，你不要被它吓到了。我们会先学习Controller是如何选举出来的，后面会再详谈Controller的具体作用。
## 课后讨论
你认为，ControllerEventManager中put方法代码是否有必要被一个Lock保护起来？
欢迎你在留言区畅所欲言，跟我交流讨论，也欢迎你把今天的内容分享给你的朋友。

550
极客时间专栏/Kafka核心源码解读/Controller模块/14 | Controller选举是怎么实现的？.md Normal file
View File

@@ -0,0 +1,550 @@
<audio id="audio" title="14 | Controller选举是怎么实现的？" controls="" preload="none"><source id="mp3" src="https://static001.geekbang.org/resource/audio/aa/18/aab8712a07a0638527505d3e3574df18.mp3"></audio>
你好，我是胡夕。
上节课，我们学习了单线程事件队列模型处理Controller事件的代码。Controller组件通过ControllerEventManager类构造了一个阻塞队列，同时配以专属的事件处理线程，实现了对各类ControllerEvent的处理。
这种设计思路既保证了多线程访问所需的线程安全，还简化了Controller端的代码结构，极大地提升了代码的可维护性。
今天，我们学习下Controller选举部分的源码。
还记得我在[第11节课](https://time.geekbang.org/column/article/235562)的案例中提到的“恢复大法”——删除ZooKeeper的/controller节点吗？当时，我们靠着这个“秘籍”涉险过关，既恢复了错误的集群状态，又避免了重启整个生产环境。
但你有没有想过，为什么删除/controller节点能够令集群元数据重新保持同步呢？如果不了解这背后的原理，我们是不敢贸然在生产环境做这种操作的。今天，我们要学习的就是这背后的一整套实现逻辑，重点关注下Controller是怎么被选举出来的。
我始终认为，只有掌握了这些知识，才算真正入门Kafka服务器端的代码了。作为Broker端最重要的组件之一，Controller在Kafka中的地位无可替代。整个Kafka集群就只有一个Controller，从某种意义上来说，它是目前Kafka这个分布式系统中唯一的“单点”。
因此，了解这个“单点”的选举触发场景，以及如何被选举出来的，对于我们后面深入理解Controller在集群中的作用非常有帮助。毕竟，Controller对外提供的一些服务也是采用了类似的实现原理。
## 概览
### ZooKeeper /controller节点
再次强调下，**在一个Kafka集群中，某段时间内只能有一台Broker被选举为Controller。随着时间的推移，可能会有不同的Broker陆续担任过Controller的角色，但是在某一时刻，Controller只能由一个Broker担任**。
那选择哪个Broker充当Controller呢？当前，Controller的选举过程依赖ZooKeeper完成。ZooKeeper除了扮演集群元数据的“真理之源”角色，还定义了/controller临时节点（Ephemeral Node），以协助完成Controller的选举。
下面这段代码展示的是一个双Broker的Kafka集群上的ZooKeeper中/controller节点：
```
{&quot;version&quot;:1,&quot;brokerid&quot;:0,&quot;timestamp&quot;:&quot;1585098432431&quot;}
cZxid = 0x1a
ctime = Wed Mar 25 09:07:12 CST 2020
mZxid = 0x1a
mtime = Wed Mar 25 09:07:12 CST 2020
pZxid = 0x1a
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x100002d3a1f0000
dataLength = 54
numChildren = 0
```
有两个地方的内容，你要重点关注一下。
- Controller Broker Id是0，表示序号为0的Broker是集群Controller。
- ephemeralOwner字段不是0x0，说明这是一个临时节点。
既然是临时节点，那么，一旦Broker与ZooKeeper的会话终止，该节点就会消失。Controller选举就依靠了这个特性。每个Broker都会监听/controller节点随时准备应聘Controller角色。下图展示了Broker与/controller节点的交互关系：
<img src="https://static001.geekbang.org/resource/image/2e/83/2e75cdbfb68c86169ec83f58e59e1283.jpg" alt="">
如图所示，集群上所有的Broker都在实时监听ZooKeeper上的这个节点。这里的“监听”有两个含义。
- **监听这个节点是否存在**。倘若发现这个节点不存在，Broker会立即“抢注”该节点，即创建/controller节点。创建成功的那个Broker，即当选为新一届的Controller。
- **监听这个节点数据是否发生了变更**。同样，一旦发现该节点的内容发生了变化，Broker也会立即启动新一轮的Controller选举。
掌握了这些基础之后，下面我们来阅读具体的源码文件：KafkaController.scala。这是一个2200行的大文件。我先向你介绍一下这个文件的大致结构，以免你陷入到一些繁枝末节中。
### 源码结构
KafkaController文件的代码结构如下图所示：
<img src="https://static001.geekbang.org/resource/image/7e/88/7e5ddb69df585b5bbbcc91336ab8f588.jpg" alt="">
整体而言，该文件大致由五部分组成。
- **选举触发器**（ElectionTrigger）：这里的选举不是指Controller选举，而是指主题分区副本的选举，即为哪些分区选择Leader副本。后面在学习副本管理器和分区管理器时，我们会讲到它。
- **KafkaController Object**：KafkaController伴生对象，仅仅定义了一些常量和回调函数类型。
- **ControllerEvent**：定义Controller事件类型。上节课我们详细学习过Controller事件以及基于事件的单线程事件队列模型。这部分的代码看着很多，但实际上都是千篇一律的。你看懂了一个事件的定义，其他的也就不在话下了。
- **各种ZooKeeper监听器**：定义ZooKeeper监听器，去监听ZooKeeper中各个节点的变更。今天，我们重点关注监听/controller节点的那个监听器。
- **KafkaController Class**：定义KafkaController类以及实际的处理逻辑。这是我们今天的重点学习对象。
接下来，我会给你重点介绍KafkaController类、ZooKeeper监听器和Controller选举这三大部分。在众多的ZooKeeper监听器中，我会详细介绍监听Controller变更的监听器，它也是我们了解Controller选举流程的核心环节。
## KafkaController类
这个类大约有1900行代码，里面定义了非常多的变量和方法。这些方法大多是处理不同Controller事件的。后面讲到选举流程的时候，我会挑一些有代表性的来介绍。我希望你能举一反三，借此吃透其他方法的代码。毕竟，它们做的事情大同小异，至少代码风格非常相似。
在学习重要的方法之前，我们必须要先掌握KafkaController类的定义。接下来，我们从4个维度来进行学习，分别是原生字段、辅助字段、各类ZooKeeper监听器字段和统计字段。
弄明白了这些字段的含义之后，再去看操作这些字段的方法，会更加有的放矢，理解起来也会更加容易。
### 原生字段
首先来看原生字段。所谓的原生字段，是指在创建一个KafkaController实例时，需要指定的字段。
先来看下KafkaController类的定义代码：
```
// 字段含义：
// config：Kafka配置信息，通过它，你能拿到Broker端所有参数的值
// zkClient：ZooKeeper客户端，Controller与ZooKeeper的所有交互均通过该属性完成
// time：提供时间服务(如获取当前时间)的工具类
// metrics：实现指标监控服务(如创建监控指标)的工具类
// initialBrokerInfo：Broker节点信息，包括主机名、端口号，所用监听器等
// initialBrokerEpoch：Broker Epoch值，用于隔离老Controller发送的请求
// tokenManager：实现Delegation token管理的工具类。Delegation token是一种轻量级的认证机制
// threadNamePrefix：Controller端事件处理线程名字前缀
class KafkaController(val config: KafkaConfig,
zkClient: KafkaZkClient,
time: Time,
metrics: Metrics,
initialBrokerInfo: BrokerInfo,
initialBrokerEpoch: Long,
tokenManager: DelegationTokenManager,
threadNamePrefix: Option[String] = None)
extends ControllerEventProcessor with Logging with KafkaMetricsGroup {
......
}
```
就像我上节课说过的，KafkaController实现了ControllerEventProcessor接口，因而也就实现了处理Controller事件的process方法。这里面比较重要的字段有3个。
- **config**：KafkaConfig类实例，里面封装了Broker端所有参数的值。
- **zkClient**：ZooKeeper客户端类，定义了与ZooKeeper交互的所有方法。
- **initialBrokerEpoch**：Controller所在Broker的Epoch值。Kafka使用它来确保Broker不会处理老Controller发来的请求。
其他字段要么是像time、metrics一样，是工具类字段，要么是像initialBrokerInfo、tokenManager字段一样，使用场景很有限，我就不展开讲了。
### 辅助字段
除了原生字段之外，KafkaController还定义了很多辅助字段，帮助实现Controller的各类功能。
我们来看一些重要的辅助字段：
```
......
// 集群元数据类，保存集群所有元数据
val controllerContext = new ControllerContext
// Controller端通道管理器类，负责Controller向Broker发送请求
var controllerChannelManager = new ControllerChannelManager(controllerContext, config, time, metrics,
stateChangeLogger, threadNamePrefix)
// 线程调度器，当前唯一负责定期执行Leader重选举
private[controller] val kafkaScheduler = new KafkaScheduler(1)
// Controller事件管理器，负责管理事件处理线程
private[controller] val eventManager = new ControllerEventManager(config.brokerId, this, time,
controllerContext.stats.rateAndTimeMetrics)
......
// 副本状态机，负责副本状态转换
val replicaStateMachine: ReplicaStateMachine = new ZkReplicaStateMachine(config, stateChangeLogger, controllerContext, zkClient,
new ControllerBrokerRequestBatch(config, controllerChannelManager, eventManager, controllerContext, stateChangeLogger))
// 分区状态机，负责分区状态转换
val partitionStateMachine: PartitionStateMachine = new ZkPartitionStateMachine(config, stateChangeLogger, controllerContext, zkClient,
new ControllerBrokerRequestBatch(config, controllerChannelManager, eventManager, controllerContext, stateChangeLogger))
// 主题删除管理器，负责删除主题及日志
val topicDeletionManager = new TopicDeletionManager(config, controllerContext, replicaStateMachine,
partitionStateMachine, new ControllerDeletionClient(this, zkClient))
......
```
其中，有7个字段是重中之重。
- **controllerContext**：集群元数据类，保存集群所有元数据。
- **controllerChannelManager**：Controller端通道管理器类，负责Controller向Broker发送请求。
- **kafkaScheduler**：线程调度器，当前唯一负责定期执行分区重平衡Leader选举。
- **eventManager**：Controller事件管理器，负责管理事件处理线程。
- **replicaStateMachine**：副本状态机，负责副本状态转换。
- **partitionStateMachine**：分区状态机，负责分区状态转换。
- **topicDeletionManager**：主题删除管理器，负责删除主题及日志。
### 各类ZooKeeper监听器
我们今天开头学到的ControllerChangeHandler仅仅是其中的一种。实际上，该类定义了很多监听器，如下所示：
```
// Controller节点ZooKeeper监听器
private val controllerChangeHandler = new ControllerChangeHandler(eventManager)
// Broker数量ZooKeeper监听器
private val brokerChangeHandler = new BrokerChangeHandler(eventManager)
// Broker信息变更ZooKeeper监听器集合
private val brokerModificationsHandlers: mutable.Map[Int, BrokerModificationsHandler] = mutable.Map.empty
// 主题数量ZooKeeper监听器
private val topicChangeHandler = new TopicChangeHandler(eventManager)
// 主题删除ZooKeeper监听器
private val topicDeletionHandler = new TopicDeletionHandler(eventManager)
// 主题分区变更ZooKeeper监听器
private val partitionModificationsHandlers: mutable.Map[String, PartitionModificationsHandler] = mutable.Map.empty
// 主题分区重分配ZooKeeper监听器
private val partitionReassignmentHandler = new PartitionReassignmentHandler(eventManager)
// Preferred Leader选举ZooKeeper监听器
private val preferredReplicaElectionHandler = new PreferredReplicaElectionHandler(eventManager)
// ISR副本集合变更ZooKeeper监听器
private val isrChangeNotificationHandler = new IsrChangeNotificationHandler(eventManager)
// 日志路径变更ZooKeeper监听器
private val logDirEventNotificationHandler = new LogDirEventNotificationHandler(eventManager)
```
我分别解释一下这些ZooKeeper监听器的作用：
- **controllerChangeHandler**：前面说过，它是监听/controller节点变更的。这种变更包括节点创建、删除以及数据变更。
- **brokerChangeHandler**：监听Broker的数量变化。
- **brokerModificationsHandlers**：监听Broker的数据变更，比如Broker的配置信息发生的变化。
- **topicChangeHandler**：监控主题数量变更。
- **topicDeletionHandler**：监听主题删除节点/admin/delete_topics的子节点数量变更。
- **partitionModificationsHandlers**：监控主题分区数据变更的监听器，比如，新增加了副本、分区更换了Leader副本。
- **partitionReassignmentHandler**：监听分区副本重分配任务。一旦发现新提交的任务，就为目标分区执行副本重分配。
- **preferredReplicaElectionHandler**：监听Preferred Leader选举任务。一旦发现新提交的任务，就为目标主题执行Preferred Leader选举。
- **isrChangeNotificationHandler**：监听ISR副本集合变更。一旦被触发，就需要获取ISR发生变更的分区列表，然后更新Controller端对应的Leader和ISR缓存元数据。
- **logDirEventNotificationHandler**：监听日志路径变更。一旦被触发，需要获取受影响的Broker列表，然后处理这些Broker上失效的日志路径。
我画了一张脑图，希望可以帮助你更高效地记住这些ZooKeeper监听器：
<img src="https://static001.geekbang.org/resource/image/8f/31/8feed623165ab6e50b31614e67498c31.jpg" alt="">
### 统计字段
最后，我们来看统计字段。
这些统计字段大多用于计算统计指标。有的监控指标甚至是非常重要的Controller监控项，比如ActiveControllerCount指标。下面，我们来了解下KafkaController都定义了哪些统计字段。这些指标的含义一目了然，非常清晰，我用注释的方式给出每个字段的含义：
```
// 当前Controller所在Broker Id
@volatile private var activeControllerId = -1
// 离线分区总数
@volatile private var offlinePartitionCount = 0
// 满足Preferred Leader选举条件的总分区数
@volatile private var preferredReplicaImbalanceCount = 0
// 总主题数
@volatile private var globalTopicCount = 0
// 总主题分区数
@volatile private var globalPartitionCount = 0
// 待删除主题数
@volatile private var topicsToDeleteCount = 0
//待删除副本数
@volatile private var replicasToDeleteCount = 0
// 暂时无法删除的主题数
@volatile private var ineligibleTopicsToDeleteCount = 0
// 暂时无法删除的副本数
@volatile private var ineligibleReplicasToDeleteCount = 0
```
好了，KafkaController类的定义我们就全部介绍完了。再次强调下，因为KafkaController类的代码很多，我强烈建议你熟练掌握这些字段的含义，因为后面的所有方法都是围绕着这些字段进行操作的。
接下来，我以Controller的选举流程为例，引出KafkaController的一些方法的实现原理。不过，在此之前，我们要学习监听Controller变更的ZooKeeper监听器：ControllerChangeHandler的源码。
## ControllerChangeHandler监听器
就像我前面说到的，KafkaController定义了十几种ZooKeeper监听器。和Controller相关的监听器是ControllerChangeHandler，用于监听Controller的变更，定义代码如下：
```
class ControllerChangeHandler(eventManager: ControllerEventManager) extends ZNodeChangeHandler {
// ZooKeeper中Controller节点路径，即/controller
override val path: String = ControllerZNode.path
// 监听/controller节点创建事件
override def handleCreation(): Unit = eventManager.put(ControllerChange)
// 监听/controller节点被删除事件
override def handleDeletion(): Unit = eventManager.put(Reelect)
// 监听/controller节点数据变更事件
override def handleDataChange(): Unit = eventManager.put(ControllerChange)
}
```
该监听器接收ControllerEventManager实例，实现了ZNodeChangeHandler接口的三个方法：**handleCreation**、**handleDeletion**和**handleDataChange**。该监听器下的path变量，实际上就是/controller字符串，表示它监听ZooKeeper的这个节点。
3个handle方法都用于监听/controller节点的变更，但实现细节上稍有不同。
handleCreation和handleDataChange的处理方式是向事件队列写入ControllerChange事件；handleDeletion的处理方式是向事件队列写入Reelect事件。
Deletion表明ZooKeeper中/controller节点不存在了，即Kafka集群中的Controller暂时空缺了。因为它和Creation和DataChange是不同的状态，需要区别对待，因此，Reelect事件做的事情要比ControllerChange的多：处理ControllerChange事件，只需要当前Broker执行“卸任Controller”的逻辑即可，而Reelect事件是重选举，除了Broker执行卸任逻辑之外，还要求Broker参与到重选举中来。
由于KafkaController的process方法代码非常长，因此，我节选了刚刚提到的那两个事件的处理代码：
```
// process方法(部分)
override def process(event: ControllerEvent): Unit = {
try {
event match {
......
// ControllerChange事件
case ControllerChange =&gt;
processControllerChange()
// Reelect事件
case Reelect =&gt;
processReelect()
......
}
}
......
}
// 如果是ControllerChange事件，仅执行卸任逻辑即可
private def processControllerChange(): Unit = {
maybeResign()
}
// 如果是Reelect事件，还需要执行elect方法参与新一轮的选举
private def processReelect(): Unit = {
maybeResign()
elect()
}
```
可以看到，虽然代码非常长，但整体结构却工整清晰，全部都是基于模式匹配的事件处理。process方法会根据给定的Controller事件类型，调用对应的process***方法处理该事件。这里只列举了ZooKeeper端/controller节点监听器监听的两类事件，以及对应的处理方法。
对于ControllerChange事件而言，处理方式是调用maybeResign去执行Controller的卸任逻辑。如果是Reelect事件，除了执行卸任逻辑之外，还要额外执行elect方法进行新一轮的Controller选举。
## Controller选举流程
说完了ControllerChangeHandler源码，我们来看下Controller的选举。所谓的Controller选举，是指Kafka选择集群中一台Broker行使Controller职责。整个选举过程分为两个步骤：触发选举和开始选举。
### 触发选举
我先用一张图展示下可能触发Controller选举的三个场景。
<img src="https://static001.geekbang.org/resource/image/a8/98/a8cbc562518f93f9befc6bd7a87d5b98.jpg" alt="">
这三个场景是：
1. 集群从零启动时；
1. Broker侦测/controller节点消失时；
1. Broker侦测到/controller节点数据发生变更时。
这三个场景殊途同归，最后都要执行选举Controller的动作。我来一一解释下这三个场景，然后再介绍选举Controller的具体操作。
#### 场景一：集群从零启动
集群首次启动时，Controller尚未被选举出来。于是，Broker启动后，首先将Startup这个ControllerEvent写入到事件队列中，然后启动对应的事件处理线程和ControllerChangeHandler ZooKeeper监听器，最后依赖事件处理线程进行Controller的选举。
在源码中，KafkaController类的startup方法就是做这些事情的。当Broker启动时，它会调用这个方法启动ControllerEventThread线程。值得注意的是，**每个Broker都需要做这些事情，不是说只有Controller所在的Broker才需要执行这些逻辑**。
startup方法的主体代码如下：
```
def startup() = {
// 第1步：注册ZooKeeper状态变更监听器，它是用于监听Zookeeper会话过期的
zkClient.registerStateChangeHandler(new StateChangeHandler {
override val name: String = StateChangeHandlers.ControllerHandler
override def afterInitializingSession(): Unit = {
eventManager.put(RegisterBrokerAndReelect)
}
override def beforeInitializingSession(): Unit = {
val queuedEvent = eventManager.clearAndPut(Expire)
queuedEvent.awaitProcessing()
}
})
// 第2步：写入Startup事件到事件队列
eventManager.put(Startup)
// 第3步：启动ControllerEventThread线程，开始处理事件队列中的ControllerEvent
eventManager.start()
}
```
首先，startup方法会注册ZooKeeper状态变更监听器，用于监听Broker与ZooKeeper之间的会话是否过期。接着，写入Startup事件到事件队列，然后启动ControllerEventThread线程，开始处理事件队列中的Startup事件。
接下来，我们来学习下KafkaController的process方法处理Startup事件的方法：
```
// KafkaController的process方法，
override def process(event: ControllerEvent): Unit = {
try {
event match {
......
case Startup =&gt;
processStartup() // 处理Startup事件
}
}
......
}
private def processStartup(): Unit = {
// 注册ControllerChangeHandler ZooKeeper监听器
zkClient.registerZNodeChangeHandlerAndCheckExistence(
controllerChangeHandler)
// 执行Controller选举
elect()
}
```
从这段代码可知，process方法调用processStartup方法去处理Startup事件。而processStartup方法又会调用zkClient的registerZNodeChangeHandlerAndCheckExistence方法注册ControllerChangeHandler监听器。
值得注意的是，虽然前面的三个场景是并列的关系，但实际上，后面的两个场景必须要等场景一的这一步成功执行之后，才能被触发。
这三种场景都要选举Controller，因此，我们最后统一学习elect方法的代码实现。
总体来说，集群启动时，Broker通过向事件队列“塞入”Startup事件的方式，来触发Controller的竞选。
#### 场景二：/controller节点消失
Broker检测到/controller节点消失时，就意味着，此时整个集群中没有Controller。因此，所有检测到/controller节点消失的Broker，都会立即调用elect方法执行竞选逻辑。
你可能会问：“Broker是怎么侦测到ZooKeeper上的这一变化的呢？”实际上，这是ZooKeeper监听器提供的功能，换句话说，这是Apache ZooKeeper自己实现的功能，所以我们才说，Kafka依赖ZooKeeper完成Controller的选举。
讲到这里，我说点题外话，社区最近正在酝酿彻底移除ZooKeeper依赖。具体到Controller端的变化，就是在Kafka内部实现一个类似于Raft的共识算法来选举Controller。我会在后面的特别放送里详细讲一下社区移除ZooKeeper的全盘计划。
#### 场景三：/controller节点数据变更
Broker检测到/controller节点数据发生变化，通常表明，Controller“易主”了，这就分为两种情况：
- 如果Broker之前是Controller，那么该Broker需要首先执行卸任操作，然后再尝试竞选；
- 如果Broker之前不是Controller，那么，该Broker直接去竞选新Controller。
具体到代码层面，maybeResign方法形象地说明了这两种情况。你要注意方法中的maybe字样，这表明，Broker可能需要执行卸任操作，也可能不需要。Kafka源码非常喜欢用maybe***来命名方法名，以表示那些在特定条件下才需要执行的逻辑。以下是maybeResign的实现：
```
private def maybeResign(): Unit = {
// 非常关键的一步！这是判断是否需要执行卸任逻辑的重要依据！
// 判断该Broker之前是否是Controller
val wasActiveBeforeChange = isActive
// 注册ControllerChangeHandler监听器
zkClient.registerZNodeChangeHandlerAndCheckExistence(
controllerChangeHandler)
// 获取当前集群Controller所在的Broker Id，如果没有Controller则返回-1
activeControllerId = zkClient.getControllerId.getOrElse(-1)
// 如果该Broker之前是Controller但现在不是了
if (wasActiveBeforeChange &amp;&amp; !isActive) {
onControllerResignation() // 执行卸任逻辑
}
}
```
代码的第一行非常关键，它是决定是否需要执行卸任的重要依据。毕竟，如果Broker之前不是Controller，那何来“卸任”一说呢？之后代码要注册ControllerChangeHandler监听器，获取当前集群Controller所在的Broker ID，如果没有Controller，则返回-1。有了这些数据之后，maybeResign方法需要判断该Broker是否之前是Controller但现在不是了。如果是这种情况的话，则调用onControllerResignation方法执行Controller卸任逻辑。
说到“卸任”，你可能会问：“卸任逻辑是由哪个方法执行的呢？”实际上，这是由onControllerResignation方法执行的，它主要是用于清空各种数据结构的值、取消ZooKeeper监听器、关闭各种状态机以及管理器，等等。我用注释的方式给出它的逻辑实现：
```
private def onControllerResignation(): Unit = {
debug(&quot;Resigning&quot;)
// 取消ZooKeeper监听器的注册
zkClient.unregisterZNodeChildChangeHandler(
isrChangeNotificationHandler.path)
zkClient.unregisterZNodeChangeHandler(
partitionReassignmentHandler.path)
zkClient.unregisterZNodeChangeHandler(
preferredReplicaElectionHandler.path)
zkClient.unregisterZNodeChildChangeHandler(
logDirEventNotificationHandler.path)
unregisterBrokerModificationsHandler(
brokerModificationsHandlers.keySet)
// 关闭Kafka线程调度器，其实就是取消定期的Leader重选举
kafkaScheduler.shutdown()
// 将统计字段全部清0
offlinePartitionCount = 0
preferredReplicaImbalanceCount = 0
globalTopicCount = 0
globalPartitionCount = 0
topicsToDeleteCount = 0
replicasToDeleteCount = 0
ineligibleTopicsToDeleteCount = 0
ineligibleReplicasToDeleteCount = 0
// 关闭Token过期检查调度器
if (tokenCleanScheduler.isStarted)
tokenCleanScheduler.shutdown()
// 取消分区重分配监听器的注册
unregisterPartitionReassignmentIsrChangeHandlers()
// 关闭分区状态机
partitionStateMachine.shutdown()
// 取消主题变更监听器的注册
zkClient.unregisterZNodeChildChangeHandler(topicChangeHandler.path)
// 取消分区变更监听器的注册
unregisterPartitionModificationsHandlers(
partitionModificationsHandlers.keys.toSeq)
// 取消主题删除监听器的注册
zkClient.unregisterZNodeChildChangeHandler(
topicDeletionHandler.path)
// 关闭副本状态机
replicaStateMachine.shutdown()
// 取消Broker变更监听器的注册
zkClient.unregisterZNodeChildChangeHandler(brokerChangeHandler.path)
// 关闭Controller通道管理器
controllerChannelManager.shutdown()
// 清空集群元数据
controllerContext.resetContext()
info(&quot;Resigned&quot;)
}
```
### 选举Controller
讲完了触发场景，接下来，我们就要学习Controller选举的源码了。前面说过了，这三种选举场景最后都会调用elect方法来执行选举逻辑。我们来看下它的实现：
```
private def elect(): Unit = {
// 第1步：获取当前Controller所在Broker的序号，如果Controller不存在，显式标记为-1
activeControllerId = zkClient.getControllerId.getOrElse(-1)
// 第2步：如果当前Controller已经选出来了，直接返回即可
if (activeControllerId != -1) {
debug(s&quot;Broker $activeControllerId has been elected as the controller, so stopping the election process.&quot;)
return
}
try {
// 第3步：注册Controller相关信息
// 主要是创建/controller节点
val (epoch, epochZkVersion) = zkClient.registerControllerAndIncrementControllerEpoch(config.brokerId)
controllerContext.epoch = epoch
controllerContext.epochZkVersion = epochZkVersion
activeControllerId = config.brokerId
info(s&quot;${config.brokerId} successfully elected as the controller. Epoch incremented to ${controllerContext.epoch} &quot; +
s&quot;and epoch zk version is now ${controllerContext.epochZkVersion}&quot;)
// 第4步：执行当选Controller的后续逻辑
onControllerFailover()
} catch {
case e: ControllerMovedException =&gt;
maybeResign()
if (activeControllerId != -1)
debug(s&quot;Broker $activeControllerId was elected as controller instead of broker ${config.brokerId}&quot;, e)
else
warn(&quot;A controller has been elected but just resigned, this will result in another round of election&quot;, e)
case t: Throwable =&gt;
error(s&quot;Error while electing or becoming controller on broker ${config.brokerId}. &quot; +
s&quot;Trigger controller movement immediately&quot;, t)
triggerControllerMove()
}
}
```
为了帮助你更好地理解这个方法，我再画一张图来进行说明：
<img src="https://static001.geekbang.org/resource/image/23/1b/2331395774956a61f37836c46d65d01b.jpg" alt="">
该方法首先检查Controller是否已经选出来了。要知道，集群中的所有Broker都要执行这些逻辑，因此，非常有可能出现某些Broker在执行elect方法时，Controller已经被选出来的情况。如果Controller已经选出来了，那么，自然也就不用再做什么了。相反地，如果Controller尚未被选举出来，那么，代码会尝试创建/controller节点去抢注Controller。
一旦抢注成功，就调用onControllerFailover方法，执行选举成功后的动作。这些动作包括注册各类ZooKeeper监听器、删除日志路径变更和ISR副本变更通知事件、启动Controller通道管理器，以及启动副本状态机和分区状态机。
如果抢注失败了，代码会抛出ControllerMovedException异常。这通常表明Controller已经被其他Broker抢先占据了，那么，此时代码调用maybeResign方法去执行卸任逻辑。
## 总结
今天，我们梳理了Controller选举的全过程，包括Controller如何借助ZooKeeper监听器实现监听Controller节点，以及Controller的选举触发场景和完整流程。我们来回顾一下这节课的重点。
- Controller依赖ZooKeeper实现Controller选举，主要是借助于/controller临时节点和ZooKeeper的监听器机制。
- Controller触发场景有3种：集群启动时；/controller节点被删除时；/controller节点数据变更时。
- 源码最终调用elect方法实现Controller选举。
<img src="https://static001.geekbang.org/resource/image/e2/74/e28c134e4fd11ff8ed87933aee88d374.jpg" alt="">
下节课，我将带你学习Controller的其他重要功能，包括它如何管理Broker和副本等。你千万不要错过。
## 课后讨论
在这节课刚开始的时候，我提到，删除/controller会触发Controller选举，之后会同步集群元数据信息。那么，你知道源码是在哪里更新的元数据请求吗？
欢迎你在留言区畅所欲言，跟我交流讨论，也欢迎你把今天的内容分享给你的朋友。

409
极客时间专栏/Kafka核心源码解读/Controller模块/15 | 如何理解Controller在Kafka集群中的作用？.md Normal file
View File

@@ -0,0 +1,409 @@
<audio id="audio" title="15 | 如何理解Controller在Kafka集群中的作用？" controls="" preload="none"><source id="mp3" src="https://static001.geekbang.org/resource/audio/5a/41/5afb796241d45b2e75d1f295067f0441.mp3"></audio>
你好，我是胡夕。
上节课，我们学习了Controller选举的源码，了解了Controller组件的选举触发场景，以及它是如何被选举出来的。Controller就绪之后，就会行使它作为控制器的重要权利了，包括管理集群成员、维护主题、操作元数据，等等。
之前在学习Kafka的时候，我一直很好奇，新启动的Broker是如何加入到集群中的。官方文档里的解释是：“Adding servers to a Kafka cluster is easy, just assign them a unique broker id and start up Kafka on your new servers.”显然，你只要启动Broker进程，就可以实现集群的扩展，甚至包括集群元数据信息的同步。
不过，你是否思考过，这一切是怎么做到的呢？其实，这就是Controller组件源码提供的一个重要功能：管理新集群成员。
当然，作为核心组件，Controller提供的功能非常多。除了集群成员管理，主题管理也是一个极其重要的功能。今天，我就带你深入了解下它们的实现代码。可以说，这是Controller最核心的两个功能，它们几乎涉及到了集群元数据中的所有重要数据。掌握了这些，之后你在探索Controller的其他代码时，会更加游刃有余。
## 集群成员管理
首先，我们来看Controller管理集群成员部分的代码。这里的成员管理包含两个方面：
1. 成员数量的管理，主要体现在新增成员和移除现有成员；
1. 单个成员的管理，如变更单个Broker的数据等。
### 成员数量管理
每个Broker在启动的时候，会在ZooKeeper的/brokers/ids节点下创建一个名为broker.id参数值的临时节点。
举个例子，假设Broker的broker.id参数值设置为1001，那么，当Broker启动后，你会在ZooKeeper的/brokers/ids下观测到一个名为1001的子节点。该节点的内容包括了Broker配置的主机名、端口号以及所用监听器的信息（注意：这里的监听器和上面说的ZooKeeper监听器不是一回事）。
当该Broker正常关闭或意外退出时，ZooKeeper上对应的临时节点会自动消失。
基于这种临时节点的机制，Controller定义了BrokerChangeHandler监听器，专门负责监听/brokers/ids下的子节点数量变化。
一旦发现新增或删除Broker，/brokers/ids下的子节点数目一定会发生变化。这会被Controller侦测到，进而触发BrokerChangeHandler的处理方法，即handleChildChange方法。
我给出BrokerChangeHandler的代码。可以看到，这里面定义了handleChildChange方法：
```
class BrokerChangeHandler(eventManager: ControllerEventManager) extends ZNodeChildChangeHandler {
// Broker ZooKeeper ZNode: /brokers/ids
override val path: String = BrokerIdsZNode.path
override def handleChildChange(): Unit = {
eventManager.put(BrokerChange) // 仅仅是向事件队列写入BrokerChange事件
}
}
```
该方法的作用就是向Controller事件队列写入一个BrokerChange事件。**事实上，Controller端定义的所有Handler的处理逻辑，都是向事件队列写入相应的ControllerEvent，真正的事件处理逻辑位于KafkaController类的process方法中。**
那么，接下来，我们就来看process方法。你会发现，处理BrokerChange事件的方法实际上是processBrokerChange，代码如下：
```
private def processBrokerChange(): Unit = {
// 如果该Broker不是Controller，自然无权处理，直接返回
if (!isActive) return
// 第1步：从ZooKeeper中获取集群Broker列表
val curBrokerAndEpochs = zkClient.getAllBrokerAndEpochsInCluster
val curBrokerIdAndEpochs = curBrokerAndEpochs map { case (broker, epoch) =&gt; (broker.id, epoch) }
val curBrokerIds = curBrokerIdAndEpochs.keySet
// 第2步：获取Controller当前保存的Broker列表
val liveOrShuttingDownBrokerIds = controllerContext.liveOrShuttingDownBrokerIds
// 第3步：比较两个列表，获取新增Broker列表、待移除Broker列表、
// 已重启Broker列表和当前运行中的Broker列表
val newBrokerIds = curBrokerIds.diff(liveOrShuttingDownBrokerIds)
val deadBrokerIds = liveOrShuttingDownBrokerIds.diff(curBrokerIds)
val bouncedBrokerIds = (curBrokerIds &amp; liveOrShuttingDownBrokerIds)
.filter(brokerId =&gt; curBrokerIdAndEpochs(brokerId) &gt; controllerContext.liveBrokerIdAndEpochs(brokerId))
val newBrokerAndEpochs = curBrokerAndEpochs.filter { case (broker, _) =&gt; newBrokerIds.contains(broker.id) }
val bouncedBrokerAndEpochs = curBrokerAndEpochs.filter { case (broker, _) =&gt; bouncedBrokerIds.contains(broker.id) }
val newBrokerIdsSorted = newBrokerIds.toSeq.sorted
val deadBrokerIdsSorted = deadBrokerIds.toSeq.sorted
val liveBrokerIdsSorted = curBrokerIds.toSeq.sorted
val bouncedBrokerIdsSorted = bouncedBrokerIds.toSeq.sorted
info(s&quot;Newly added brokers: ${newBrokerIdsSorted.mkString(&quot;,&quot;)}, &quot; +
s&quot;deleted brokers: ${deadBrokerIdsSorted.mkString(&quot;,&quot;)}, &quot; +
s&quot;bounced brokers: ${bouncedBrokerIdsSorted.mkString(&quot;,&quot;)}, &quot; +
s&quot;all live brokers: ${liveBrokerIdsSorted.mkString(&quot;,&quot;)}&quot;)
// 第4步：为每个新增Broker创建与之连接的通道管理器和
// 底层的请求发送线程（RequestSendThread）
newBrokerAndEpochs.keySet.foreach(
controllerChannelManager.addBroker)
// 第5步：为每个已重启的Broker移除它们现有的配套资源
//（通道管理器、RequestSendThread等），并重新添加它们
bouncedBrokerIds.foreach(controllerChannelManager.removeBroker)
bouncedBrokerAndEpochs.keySet.foreach(
controllerChannelManager.addBroker)
// 第6步：为每个待移除Broker移除对应的配套资源
deadBrokerIds.foreach(controllerChannelManager.removeBroker)
// 第7步：为新增Broker执行更新Controller元数据和Broker启动逻辑
if (newBrokerIds.nonEmpty) {
controllerContext.addLiveBrokers(newBrokerAndEpochs)
onBrokerStartup(newBrokerIdsSorted)
}
// 第8步：为已重启Broker执行重添加逻辑，包含
// 更新ControllerContext、执行Broker重启动逻辑
if (bouncedBrokerIds.nonEmpty) {
controllerContext.removeLiveBrokers(bouncedBrokerIds)
onBrokerFailure(bouncedBrokerIdsSorted)
controllerContext.addLiveBrokers(bouncedBrokerAndEpochs)
onBrokerStartup(bouncedBrokerIdsSorted)
}
// 第9步：为待移除Broker执行移除ControllerContext和Broker终止逻辑
if (deadBrokerIds.nonEmpty) {
controllerContext.removeLiveBrokers(deadBrokerIds)
onBrokerFailure(deadBrokerIdsSorted)
}
if (newBrokerIds.nonEmpty || deadBrokerIds.nonEmpty ||
bouncedBrokerIds.nonEmpty) {
info(s&quot;Updated broker epochs cache: ${controllerContext.liveBrokerIdAndEpochs}&quot;)
}
}
```
代码有点长，你可以看下我添加的重点注释。同时，我再画一张图，帮你梳理下这个方法做的事情。
<img src="https://static001.geekbang.org/resource/image/ff/d3/fffc8456d8ede9219462e607fa4241d3.jpg" alt="">
整个方法共有9步。
第1~3步：
前两步分别是从ZooKeeper和ControllerContext中获取Broker列表；第3步是获取4个Broker列表：新增Broker列表、待移除Broker列表、已重启的Broker列表和当前运行中的Broker列表。
假设前两步中的Broker列表分别用A和B表示，由于Kafka以ZooKeeper上的数据为权威数据，因此，A就是最新的运行中Broker列表，“A-B”就表示新增的Broker，“B-A”就表示待移除的Broker。
已重启的Broker的判断逻辑要复杂一些，它判断的是A∧B集合中的那些Epoch值变更了的Broker。你大体上可以把Epoch值理解为Broker的版本或重启的次数。显然，Epoch值变更了，就说明Broker发生了重启行为。
第4~9步：
拿到这些集合之后，Controller会分别为这4个Broker列表执行相应的操作，也就是这个方法中第4~9步要做的事情。总体上，这些相应的操作分为3类。
- 执行元数据更新操作：调用ControllerContext类的各个方法，更新不同的集群元数据信息。比如需要将新增Broker加入到集群元数据，将待移除Broker从元数据中移除等。
- 执行Broker终止操作：为待移除Broker和已重启Broker调用onBrokerFailure方法。
- 执行Broker启动操作：为已重启Broker和新增Broker调用onBrokerStartup方法。
下面我们深入了解下onBrokerFailure和onBrokerStartup方法的逻辑。相比于其他方法，这两个方法的代码逻辑有些复杂，要做的事情也很多，因此，我们重点研究下它们。
首先是处理Broker终止逻辑的onBrokerFailure方法，代码如下：
```
private def onBrokerFailure(deadBrokers: Seq[Int]): Unit = {
info(s&quot;Broker failure callback for ${deadBrokers.mkString(&quot;,&quot;)}&quot;)
// 第1步：为每个待移除Broker，删除元数据对象中的相关项
deadBrokers.foreach(controllerContext.replicasOnOfflineDirs.remove
// 第2步：将待移除Broker从元数据对象中处于已关闭状态的Broker列表中去除
val deadBrokersThatWereShuttingDown =
deadBrokers.filter(id =&gt; controllerContext.shuttingDownBrokerIds.remove(id))
if (deadBrokersThatWereShuttingDown.nonEmpty)
info(s&quot;Removed ${deadBrokersThatWereShuttingDown.mkString(&quot;,&quot;)} from list of shutting down brokers.&quot;)
// 第3步：找出待移除Broker上的所有副本对象，执行相应操作，
// 将其置为“不可用状态”（即Offline）
val allReplicasOnDeadBrokers = controllerContext.replicasOnBrokers(deadBrokers.toSet)
onReplicasBecomeOffline(allReplicasOnDeadBrokers)
// 第4步：注销注册的BrokerModificationsHandler监听器
unregisterBrokerModificationsHandler(deadBrokers)
}
```
Broker终止，意味着我们必须要删除Controller元数据缓存中与之相关的所有项，还要处理这些Broker上保存的副本。最后，我们还要注销之前为该Broker注册的BrokerModificationsHandler监听器。
其实，主体逻辑在于如何处理Broker上的副本对象，即onReplicasBecomeOffline方法。该方法大量调用了Kafka副本管理器和分区管理器的相关功能，后面我们会专门学习这两个管理器，因此这里我就不展开讲了。
现在，我们看onBrokerStartup方法。它是处理Broker启动的方法，也就是Controller端应对集群新增Broker启动的方法。同样，我先给出带注释的完整方法代码：
```
private def onBrokerStartup(newBrokers: Seq[Int]): Unit = {
info(s&quot;New broker startup callback for ${newBrokers.mkString(&quot;,&quot;)}&quot;)
// 第1步：移除元数据中新增Broker对应的副本集合
newBrokers.foreach(controllerContext.replicasOnOfflineDirs.remove)
val newBrokersSet = newBrokers.toSet
val existingBrokers = controllerContext.
liveOrShuttingDownBrokerIds.diff(newBrokersSet)
// 第2步：给集群现有Broker发送元数据更新请求，令它们感知到新增Broker的到来
sendUpdateMetadataRequest(existingBrokers.toSeq, Set.empty)
// 第3步：给新增Broker发送元数据更新请求，令它们同步集群当前的所有分区数据
sendUpdateMetadataRequest(newBrokers, controllerContext.partitionLeadershipInfo.keySet)
val allReplicasOnNewBrokers = controllerContext.replicasOnBrokers(newBrokersSet)
// 第4步：将新增Broker上的所有副本设置为Online状态，即可用状态
replicaStateMachine.handleStateChanges(
allReplicasOnNewBrokers.toSeq, OnlineReplica)
partitionStateMachine.triggerOnlinePartitionStateChange()
// 第5步：重启之前暂停的副本迁移操作
maybeResumeReassignments { (_, assignment) =&gt;
assignment.targetReplicas.exists(newBrokersSet.contains)
}
val replicasForTopicsToBeDeleted = allReplicasOnNewBrokers.filter(p =&gt; topicDeletionManager.isTopicQueuedUpForDeletion(p.topic))
// 第6步：重启之前暂停的主题删除操作
if (replicasForTopicsToBeDeleted.nonEmpty) {
info(s&quot;Some replicas ${replicasForTopicsToBeDeleted.mkString(&quot;,&quot;)} for topics scheduled for deletion &quot; +
s&quot;${controllerContext.topicsToBeDeleted.mkString(&quot;,&quot;)} are on the newly restarted brokers &quot; +
s&quot;${newBrokers.mkString(&quot;,&quot;)}. Signaling restart of topic deletion for these topics&quot;)
topicDeletionManager.resumeDeletionForTopics(
replicasForTopicsToBeDeleted.map(_.topic))
}
// 第7步：为新增Broker注册BrokerModificationsHandler监听器
registerBrokerModificationsHandler(newBrokers)
}
```
如代码所示，第1步是移除新增Broker在元数据缓存中的信息。你可能会问：“这些Broker不都是新增的吗？元数据缓存中有它们的数据吗？”实际上，这里的newBrokers仅仅表示新启动的Broker，它们不一定是全新的Broker。因此，这里的删除元数据缓存是非常安全的做法。
第2、3步：分别给集群的已有Broker和新增Broker发送更新元数据请求。这样一来，整个集群上的Broker就可以互相感知到彼此，而且最终所有的Broker都能保存相同的分区数据。
第4步：将新增Broker上的副本状态置为Online状态。Online状态表示这些副本正常提供服务，即Leader副本对外提供读写服务，Follower副本自动向Leader副本同步消息。
第5、6步：分别重启可能因为新增Broker启动、而能够重新被执行的副本迁移和主题删除操作。
第7步：为所有新增Broker注册BrokerModificationsHandler监听器，允许Controller监控它们在ZooKeeper上的节点的数据变更。
### 成员信息管理
了解了Controller管理集群成员数量的机制之后，接下来，我们要重点学习下Controller如何监听Broker端信息的变更，以及具体的操作。
和管理集群成员类似，Controller也是通过ZooKeeper监听器的方式来应对Broker的变化。这个监听器就是BrokerModificationsHandler。一旦Broker的信息发生变更，该监听器的handleDataChange方法就会被调用，向事件队列写入BrokerModifications事件。
KafkaController类的processBrokerModification方法负责处理这类事件，代码如下：
```
private def processBrokerModification(brokerId: Int): Unit = {
if (!isActive) return
// 第1步：获取目标Broker的详细数据，
// 包括每套监听器配置的主机名、端口号以及所使用的安全协议等
val newMetadataOpt = zkClient.getBroker(brokerId)
// 第2步：从元数据缓存中获得目标Broker的详细数据
val oldMetadataOpt = controllerContext.liveOrShuttingDownBroker(brokerId)
if (newMetadataOpt.nonEmpty &amp;&amp; oldMetadataOpt.nonEmpty) {
val oldMetadata = oldMetadataOpt.get
val newMetadata = newMetadataOpt.get
// 第3步：如果两者不相等，说明Broker数据发生了变更
// 那么，更新元数据缓存，以及执行onBrokerUpdate方法处理Broker更新的逻辑
if (newMetadata.endPoints != oldMetadata.endPoints) {
info(s&quot;Updated broker metadata: $oldMetadata -&gt; $newMetadata&quot;)
controllerContext.updateBrokerMetadata(oldMetadata, newMetadata)
onBrokerUpdate(brokerId)
}
}
}
```
该方法首先获取ZooKeeper上最权威的Broker数据，将其与元数据缓存上的数据进行比对。如果发现两者不一致，就会更新元数据缓存，同时调用onBrokerUpdate方法执行更新逻辑。
那么，onBrokerUpdate方法又是如何实现的呢？我们先看下代码：
```
private def onBrokerUpdate(updatedBrokerId: Int): Unit = {
info(s&quot;Broker info update callback for $updatedBrokerId&quot;)
// 给集群所有Broker发送UpdateMetadataRequest，让她它们去更新元数据
sendUpdateMetadataRequest(
controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set.empty)
}
```
可以看到，onBrokerUpdate就是向集群所有Broker发送更新元数据信息请求，把变更信息广播出去。
怎么样，应对Broker信息变更的方法还是比较简单的吧？
## 主题管理
除了维护集群成员之外，Controller还有一个重要的任务，那就是对所有主题进行管理，主要包括主题的创建、变更与删除。
掌握了前面集群成员管理的方法，在学习下面的内容时会轻松很多。因为它们的实现机制是一脉相承的，几乎没有任何差异。
### 主题创建/变更
我们重点学习下主题是如何被创建的。实际上，主题变更与创建是相同的逻辑，因此，源码使用了一套监听器统一处理这两种情况。
你一定使用过Kafka的kafka-topics脚本或AdminClient创建主题吧？实际上，这些工具仅仅是向ZooKeeper对应的目录下写入相应的数据而已，那么，Controller，或者说Kafka集群是如何感知到新创建的主题的呢？
这当然要归功于监听主题路径的ZooKeeper监听器：TopicChangeHandler。代码如下：
```
class TopicChangeHandler(eventManager: ControllerEventManager) extends ZNodeChildChangeHandler {
// ZooKeeper节点：/brokers/topics
override val path: String = TopicsZNode.path
// 向事件队列写入TopicChange事件
override def handleChildChange(): Unit = eventManager.put(TopicChange)
}
```
代码中的TopicsZNode.path就是ZooKeeper下/brokers/topics节点。一旦该节点下新增了主题信息，该监听器的handleChildChange就会被触发，Controller通过ControllerEventManager对象，向事件队列写入TopicChange事件。
KafkaController的process方法接到该事件后，调用processTopicChange方法执行主题创建。代码如下：
```
private def processTopicChange(): Unit = {
if (!isActive) return
// 第1步：从ZooKeeper中获取所有主题
val topics = zkClient.getAllTopicsInCluster(true)
// 第2步：与元数据缓存比对，找出新增主题列表与已删除主题列表
val newTopics = topics -- controllerContext.allTopics
val deletedTopics = controllerContext.allTopics.diff(topics)
// 第3步：使用ZooKeeper中的主题列表更新元数据缓存
controllerContext.setAllTopics(topics)
// 第4步：为新增主题注册分区变更监听器
// 分区变更监听器是监听主题分区变更的
registerPartitionModificationsHandlers(newTopics.toSeq)
// 第5步：从ZooKeeper中获取新增主题的副本分配情况
val addedPartitionReplicaAssignment = zkClient.getFullReplicaAssignmentForTopics(newTopics)
// 第6步：清除元数据缓存中属于已删除主题的缓存项
deletedTopics.foreach(controllerContext.removeTopic)
// 第7步：为新增主题更新元数据缓存中的副本分配条目
addedPartitionReplicaAssignment.foreach {
case (topicAndPartition, newReplicaAssignment) =&gt; controllerContext.updatePartitionFullReplicaAssignment(topicAndPartition, newReplicaAssignment)
}
info(s&quot;New topics: [$newTopics], deleted topics: [$deletedTopics], new partition replica assignment &quot; +
s&quot;[$addedPartitionReplicaAssignment]&quot;)
// 第8步：调整新增主题所有分区以及所属所有副本的运行状态为“上线”状态
if (addedPartitionReplicaAssignment.nonEmpty)
onNewPartitionCreation(addedPartitionReplicaAssignment.keySet)
}
```
虽然一共有8步，但大部分的逻辑都与更新元数据缓存项有关，因此，处理逻辑总体上还是比较简单的。需要注意的是，第8步涉及到了使用分区管理器和副本管理器来调整分区和副本状态。后面我们会详细介绍。这里你只需要知道，分区和副本处于“上线”状态，就表明它们能够正常工作，就足够了。
### 主题删除
和主题创建或变更类似，删除主题也依赖ZooKeeper监听器完成。
Controller定义了TopicDeletionHandler，用它来实现对删除主题的监听，代码如下：
```
class TopicDeletionHandler(eventManager: ControllerEventManager) extends ZNodeChildChangeHandler {
// ZooKeeper节点：/admin/delete_topics
override val path: String = DeleteTopicsZNode.path
// 向事件队列写入TopicDeletion事件
override def handleChildChange(): Unit = eventManager.put(TopicDeletion)
}
```
这里的DeleteTopicsZNode.path指的是/admin/delete_topics节点。目前，无论是kafka-topics脚本，还是AdminClient，删除主题都是在/admin/delete_topics节点下创建名为待删除主题名的子节点。
比如，如果我要删除test-topic主题，那么，Kafka的删除命令仅仅是在ZooKeeper上创建/admin/delete_topics/test-topic节点。一旦监听到该节点被创建，TopicDeletionHandler的handleChildChange方法就会被触发，Controller会向事件队列写入TopicDeletion事件。
处理TopicDeletion事件的方法是processTopicDeletion，代码如下：
```
private def processTopicDeletion(): Unit = {
if (!isActive) return
// 从ZooKeeper中获取待删除主题列表
var topicsToBeDeleted = zkClient.getTopicDeletions.toSet
debug(s&quot;Delete topics listener fired for topics ${topicsToBeDeleted.mkString(&quot;,&quot;)} to be deleted&quot;)
// 找出不存在的主题列表
val nonExistentTopics = topicsToBeDeleted -- controllerContext.allTopics
if (nonExistentTopics.nonEmpty) {
warn(s&quot;Ignoring request to delete non-existing topics ${nonExistentTopics.mkString(&quot;,&quot;)}&quot;)
zkClient.deleteTopicDeletions(nonExistentTopics.toSeq, controllerContext.epochZkVersion)
}
topicsToBeDeleted --= nonExistentTopics
// 如果delete.topic.enable参数设置成true
if (config.deleteTopicEnable) {
if (topicsToBeDeleted.nonEmpty) {
info(s&quot;Starting topic deletion for topics ${topicsToBeDeleted.mkString(&quot;,&quot;)}&quot;)
topicsToBeDeleted.foreach { topic =&gt;
val partitionReassignmentInProgress = controllerContext.partitionsBeingReassigned.map(_.topic).contains(topic)
if (partitionReassignmentInProgress)
topicDeletionManager.markTopicIneligibleForDeletion(
Set(topic), reason = &quot;topic reassignment in progress&quot;)
}
// 将待删除主题插入到删除等待集合交由TopicDeletionManager处理
topicDeletionManager.enqueueTopicsForDeletion(topicsToBeDeleted)
}
} else { // 不允许删除主题
info(s&quot;Removing $topicsToBeDeleted since delete topic is disabled&quot;)
// 清除ZooKeeper下/admin/delete_topics下的子节点
zkClient.deleteTopicDeletions(topicsToBeDeleted.toSeq, controllerContext.epochZkVersion)
}
}
```
为了帮助你更直观地理解，我再画一张图来说明下：
<img src="https://static001.geekbang.org/resource/image/97/c9/976d35f7771f4cd5ef94eda856fb53c9.jpg" alt="">
首先，代码从ZooKeeper的/admin/delete_topics下获取子节点列表，即待删除主题列表。
之后，比对元数据缓存中的主题列表，获知压根不存在的主题列表。如果确实有不存在的主题，删除/admin/delete_topics下对应的子节点就行了。同时，代码会更新待删除主题列表，将这些不存在的主题剔除掉。
接着，代码会检查Broker端参数delete.topic.enable的值。如果该参数为false，即不允许删除主题，代码就会清除ZooKeeper下的对应子节点，不会做其他操作。反之，代码会遍历待删除主题列表，将那些正在执行分区迁移的主题暂时设置成“不可删除”状态。
最后，把剩下可以删除的主题交由TopicDeletionManager，由它执行真正的删除逻辑。
这里的TopicDeletionManager是Kafka专门负责删除主题的管理器，下节课我会详细讲解它的代码实现。
## 总结
今天，我们学习了Controller的两个主要功能：管理集群Broker成员和主题。这两个功能是Controller端提供的重要服务。我建议你仔细地查看这两部分的源码，弄明白Controller是如何管理集群中的重要资源的。
针对这些内容，我总结了几个重点，希望可以帮助你更好地理解和记忆。
- 集群成员管理：Controller负责对集群所有成员进行有效管理，包括自动发现新增Broker、自动处理下线Broker，以及及时响应Broker数据的变更。
- 主题管理：Controller负责对集群上的所有主题进行高效管理，包括创建主题、变更主题以及删除主题，等等。对于删除主题而言，实际的删除操作由底层的TopicDeletionManager完成。
<img src="https://static001.geekbang.org/resource/image/00/37/0035a579a02def8f5234831bf0857f37.jpg" alt="">
接下来，我们将进入到下一个模块：状态机模块。在该模块中，我们将系统学习Kafka提供的三大状态机或管理器。Controller非常依赖这些状态机对下辖的所有Kafka对象进行管理。在下一个模块中，我将带你深入了解分区或副本在底层的状态流转是怎么样的，你一定不要错过。
## 课后讨论
如果我们想要使用脚本命令增加一个主题的分区，你知道应该用KafkaController类中的哪个方法吗？
欢迎你在留言区畅所欲言，跟我交流讨论，也欢迎你把今天的内容分享给你的朋友。