单线程事件处理器ControllerEventManager

0 前言

单线程事件处理器，Controller端定义的一个组件。该组件内置了一个专属线程，负责处理其他线程发送过来的Controller事件。还定义了一些管理方法，为专属线程输送待处理事件。

0.11.0.0版本前，Controller组件源码复杂。集群元数据信息在程序中同时被多个线程访问，因此，源码里有大量Monitor锁、Lock锁或其他线程安全机制，导致晦涩难懂，改动困难
0.11.0.0版本开始，社区重构Controller代码结构。将多线程并发访问改为单线程的事件队列方式。并非说Controller只有一个线程，而是指对局部状态的访问限制在一个专属线程，即让这个特定线程排他操作Controller元数据信息

这就无需担心多线程访问引发的各种线程安全问题，简化Controller端代码。

1 基本概念

Controller单线程事件队列处理模型及其基础组件。
单线程事件处理器ControllerEventManager Controller端多线程向事件队列写入不同种类事件，如zk端注册的Watcher线程、KafkaRequestHandler线程、Kafka定时任务线程等。

事件队列的另一端，只有一个名为ControllerEventThread的线程，负责“消费”或处理队列中的事件。

即单线程事件队列模型。

2 相关类

2.1 ControllerEventProcessor

Controller端的事件处理器接口：
单线程事件处理器ControllerEventManager

API

process

接收一个Controller事件，并进行普通处理。实现Controller事件处理的主要方法

preempt

接收一个Controller事件，并抢占队列之前的Controller事件，进行优先处理。Kafka用其实现某些高优先级事件的抢占处理，目前在源码中也只有两类事件（ShutdownEventThread和Expire）需抢占式处理

KafkaController类是Controller组件的功能实现类，ControllerEventProcessor的唯一实现类。
单线程事件处理器ControllerEventManager

2.2 ControllerEvent

Controller事件，即事件队列中被处理的对象，对应ControllerEvent接口
单线程事件处理器ControllerEventManager
每个ControllerEvent都定义了一个状态。Controller处理具体事件时，会更新状态。
状态由ControllerState定义：

每类ControllerState都定义一个value值，表示Controller状态的序号，从0开始。
rateAndTimeMetricName方法用于构造Controller状态速率的监控指标名称。

如TopicChange是一类ControllerState：主题总数发生变化。

监控这类状态变更速率，rateAndTimeMetricName方法会定义名为TopicChangeRateAndTimeMs的指标。
并非所有ControllerState都有对应速率监控指标，如表示空闲状态的Idle无对应指标。

Controller总共定义了25类事件和17种状态：
单线程事件处理器ControllerEventManager

监控到某些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类：
单线程事件处理器ControllerEventManager
继承自ShutdownableThread：

该类会循环执行doWork，具体实现则由子类完成。

作为Controller唯一的事件处理线程，需关注该线程运行状态。必须要知道该线程在JVM上的名字，后续就能对其监控。
线程名由ControllerEventManager Object中ControllerEventThreadName变量定义：
单线程事件处理器ControllerEventManager
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：事件管理器创建的事件处理线程。该线程排他性地读取事件队列并处理队列中的所有事件。

秒客网