1. Johngo学长首页
  2. 技术点详细复盘专题
  3. Python

单线程事件处理器ControllerEventManager

• Python • 阅读 48

0 前言

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

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

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

1 基本概念

Controller单线程事件队列处理模型及其基础组件。

单线程事件处理器ControllerEventManager

事件队列的另一端，只有一个名为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定义：
单线程事件处理器ControllerEventManager
每类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
    保存一些字符串常量，如线程名称
    单线程事件处理器ControllerEventManager
  • ControllerEventProcessor
    事件处理器接口，目前只有KafkaController实现该接口
  • QueuedEvent
    事件队列上的事件对象
  • ControllerEventManager Class
    ControllerEventManager的伴生类，主要创建和管理事件处理线程和事件队列。该类定义了ControllerEventThread线程类。

ControllerEventManager对象仅定义了3个公共变量。

; QueuedEvent

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：
单线程事件处理器ControllerEventManager
单线程事件处理器ControllerEventManager
该类会循环执行doWork，具体实现则由子类完成。

作为Controller唯一的事件处理线程，需关注该线程运行状态。必须要知道该线程在JVM上的名字，后续就能对其监控。
线程名由ControllerEventManager Object中ControllerEventThreadName变量定义：

单线程事件处理器ControllerEventManager
ControllerEventThread#doWork 
override def doWork(): Unit = {

  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()

  processor.process(event)
}

方法首先判断该事件是否已被处理:

  • 是，直接返回
  • 不是，调用ControllerEventProcessor#process处理事件

每个ControllerEventProcessor#process都封装在KafkaController.scala文件。就是KafkaController类实现ControllerEventProcessor#process，部分代码：

override def process(event: ControllerEvent): Unit = {
    try {

      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 {

      case e: ControllerMovedException =>
        info(s"Controller moved to another broker when processing $event.", e)

        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先执行高优先级的抢占式事件，之后清空队列所有事件，最后再插入指定事件

下面这两段源码分别对应这两个方法：


def put(event: ControllerEvent): QueuedEvent = inLock(putLock) {

  val queuedEvent = new QueuedEvent(event, time.milliseconds())

  queue.put(queuedEvent)

  queuedEvent
}

def clearAndPut(event: ControllerEvent): QueuedEvent = inLock(putLock) {

  queue.forEach(_.preempt(processor))

  queue.clear()

  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：事件管理器创建的事件处理线程。该线程排他性地读取事件队列并处理队列中的所有事件。

Original: https://blog.csdn.net/qq_33589510/article/details/119834544
Author: JavaEdge.
Title: 单线程事件处理器ControllerEventManager

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/814189/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球