0 前言
单线程事件处理器，Controller端定义的一个组件。该组件内置了一个专属线程，负责处理其他线程发送过来的Controller事件。还定义了一些管理方法，为专属线程输送待处理事件。
- 0.11.0.0版本前，Controller组件源码复杂。集群元数据信息在程序中同时被多个线程访问，因此，源码里有大量Monitor锁、Lock锁或其他线程安全机制，导致晦涩难懂，改动困难
- 0.11.0.0版本开始，社区重构Controller代码结构。将多线程并发访问改为单线程的事件队列方式。并非说Controller只有一个线程，而是指对局部状态的访问限制在一个专属线程，即让这个特定线程排他操作Controller元数据信息
这就无需担心多线程访问引发的各种线程安全问题，简化Controller端代码。
1 基本概念
Controller单线程事件队列处理模型及其基础组件。
Controller端多线程向事件队列写入不同种类事件，如zk端注册的Watcher线程、KafkaRequestHandler线程、Kafka定时任务线程等。
事件队列的另一端，只有一个名为ControllerEventThread的线程，负责“消费”或处理队列中的事件。
即单线程事件队列模型。
2 相关类
2.1 ControllerEventProcessor
Controller端的事件处理器接口：
API
process
接收一个Controller事件，并进行普通处理。实现Controller事件处理的主要方法
preempt
接收一个Controller事件，并抢占队列之前的Controller事件，进行优先处理。Kafka用其实现某些高优先级事件的抢占处理，目前在源码中也只有两类事件（ShutdownEventThread和Expire）需抢占式处理
KafkaController类是Controller组件的功能实现类，ControllerEventProcessor的唯一实现类。
2.2 ControllerEvent
Controller事件，即事件队列中被处理的对象，对应ControllerEvent接口
每个ControllerEvent都定义了一个状态。Controller处理具体事件时，会更新状态。
状态由ControllerState定义：
每类ControllerState都定义一个value值，表示Controller状态的序号，从0开始。
rateAndTimeMetricName方法用于构造Controller状态速率的监控指标名称。
如TopicChange是一类ControllerState：主题总数发生变化。
监控这类状态变更速率，rateAndTimeMetricName方法会定义名为TopicChangeRateAndTimeMs的指标。
并非所有ControllerState都有对应速率监控指标，如表示空闲状态的Idle无对应指标。
Controller总共定义了25类事件和17种状态：
监控到某些Controller状态变更速率异常时，可通过该表，快速确定可能造成瓶颈的Controller事件，并定位处理函数，辅助排查。
多个ControllerEvent可能属相同的ControllerState。
2.3 ControllerEventManager
事件处理器，创建和管理ControllerEventThread。位于ControllerEventManager.scala，该文件的组成：
- ControllerEventManager Object
保存一些字符串常量，如线程名称
- ControllerEventProcessor
事件处理器接口，目前只有KafkaController实现该接口 - QueuedEvent
事件队列上的事件对象 - ControllerEventManager Class
ControllerEventManager的伴生类，主要创建和管理事件处理线程和事件队列。该类定义了ControllerEventThread线程类。
ControllerEventManager对象仅定义了3个公共变量。
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"QueuedEvent(event=$event, enqueueTimeMs=$enqueueTimeMs)"
}
}
每个QueuedEvent对象实例都裹挟了一个ControllerEvent。
在QueuedEvent中，用CountDownLatch来做各种条件控制，比如用于侦测线程是否成功启动、成功关闭等。
QueuedEvent使用它的唯一目的，是确保Expire事件在建立ZooKeeper会话前被处理。
若不是在该场景，则代码就用spent来标识该事件是否已被处理：
- 若已被处理
再次调用process方法时就会直接返回
2.4 ControllerEventThread
专属的事件处理线程，唯一作用：处理不同种类的ControllEvent。
ControllerEventManager类内部定义的线程类。
消费QueuedEvent的ControllerEventThread类：
继承自ShutdownableThread：
该类会循环执行doWork，具体实现则由子类完成。
作为Controller唯一的事件处理线程，需关注该线程运行状态。必须要知道该线程在JVM上的名字，后续就能对其监控。
线程名由ControllerEventManager Object中ControllerEventThreadName变量定义：
ControllerEventThread#doWork
override def doWork(): Unit = {
// 从事件队列获取待处理的Controller事件（QueuedEvent对象实例），否则等待
val dequeued = queue.take()
dequeued.event match {
// 若是关闭线程事件，啥都不做。关闭线程由外部来执行
case ShutdownEventThread =>
case controllerEvent =>
_state = controllerEvent.state
// 更新对应事件在队列中保存的时间
eventQueueTimeHist.update(time.milliseconds() - dequeued.enqueueTimeMs)
try {
def process(): Unit = dequeued.process(processor)
// 处理事件，同时计算处理速率
rateAndTimeMetrics.get(state) match {
case Some(timer) => timer.time { process() }
case None => process()
}
} catch {
case e: Throwable => error(s"Uncaught error processing event $controllerEvent", e)
}
_state = ControllerState.Idle
}
}
首先是调用LinkedBlockingQueue#take，去。注意，这里用的是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#process，部分代码：
override def process(event: ControllerEvent): Unit = {
try {
// 依次匹配ControllerEvent事件
event match {
case event: MockEvent =>
event.process()
case ShutdownEventThread =>
error("Received a ShutdownEventThread event. This type of event is supposed to be handle by ControllerEventThread")
case AutoPreferredReplicaLeaderElection =>
processAutoPreferredReplicaLeaderElection()
......
}
} catch {
// 如果Controller换成了别的Broker
case e: ControllerMovedException =>
info(s"Controller moved to another broker when processing $event.", e)
// 执行Controller卸任逻辑
maybeResign()
case e: Throwable =>
error(s"Error processing event $event", e)
} finally {
updateMetrics()
}
}
这个process方法接收一个ControllerEvent实例，接着会判断它是哪类Controller事件，并调用相应的处理方法：
- AutoPreferredReplicaLeaderElection，调processAutoPreferredReplicaLeaderElection
- 其他类型事件，调用process
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中使用锁，毕竟要保证，访问抢占式事件和清空操作构成一个原子操作。
3 总结
Controller端的单线程事件队列实现方式，即ControllerEventManager通过构建ControllerEvent、ControllerState和对应的ControllerEventThread线程，并且结合专属事件队列，共同实现事件处理。
ControllerEvent：定义Controller能够处理的各类事件名称，目前总共定义了25类事件。
ControllerState：定义Controller状态。ControllerEvent的上一级分类，因此，ControllerEvent和ControllerState是多对一。
ControllerEventManager：Controller定义的事件管理器，专门定义和维护专属线程以及对应的事件队列。
ControllerEventThread：事件管理器创建的事件处理线程。该线程排他性地读取事件队列并处理队列中的所有事件。
