系列文章目录

Kafka源码分析-目录

一. 背景

Kafka内部涉及大量的"延时"操作,比如收到PRODUCE请求后可为副本等待一个timeout的时间后再响应客户端。

那我们讨论一个问题:Kafka为什么自己实现了一个延时任务组件,而不直接使用java的java.util.concurrent.DelayQueue?

可以按如下两点来分析这个事:

  1. DelayQueue提供了哪些能力?

  2. Kafka所面对的场景是怎样的,为什么DelayQueue不适用?

1.1 DelayQueue的能力

DelayQueue相关的接口/类如下图:
Kafka源码分析(五) - Server端 - 基于时间轮的延时组件-LMLPHP
对应地,DelayQueue所提供的能力如下:

  1. 为任务指定延时时间
    DelayQueue所维护的任务都要实现Delayed接口;其中的getDelay方法返回了距离该任务所设定执行时间有多远;

  2. 任务存取的时间复杂度为O(log n)
    DelayQueue内部使用"优先级队列"来保存任务,该数据结构存取的时间复杂度为对数复杂度;

1.2 Kafka的业务场景

Kafka的业务背景有如下特点:

  1. 延时任务不一定非得等到超时后才执行,有一些事件会触发其提前执行;
    比如PRODUCE请求处理过程中,若副本的响应比较快,那收到了预期数量的副本响应后就可以开始给客户端发响应,不一定非得等满一个timeout的时间;

  2. 延时任务的"入队"操作QPS很高;
    Kafka就是为高QPS而生,内部会产生大量的延时任务;

对应地,Kafka对延时任务组件有如下两点要求:

  1. 要求有提前执行任务的能力;

  2. 要求尽可能降低延时任务入队操作的时间复杂度;
    (借助一个名为"时间轮"的数据结构,Kafka将时间复杂度实际降到了O(1))

这两点要求都无法通过直接应用DelayQueue的方式得到满足。

二. 组件接口

我们来看看Kafka的延时任务组件对外提供的接口,进而搞清楚其提供的能力和使用方式。

如下图:
Kafka源码分析(五) - Server端 - 基于时间轮的延时组件-LMLPHP
左边两个类定义了"延时操作":

  1. TimerTask
    描述了一个最基本的延时Task;定义了超时时间(delayMs属性)、提供了一个取消操作(cancel方法);

  2. DelayedOperation
    描述了一个可被外部事件提前触发的TimerTask;其在TimerTask基础上提供了检查是否满足提前触发条件的方法(tryComplete)、并定义被外部事件提前触发后的行为(onComplete)和无事件触发直到超时后的行为(onExpiration);这几个方法都需要DelayedOperation的子类进行实现;

右边的DelayedOperationPurgatory类定义了一个维护DelayOperaton的容器,其核心操作如下:

  1. tryCompleteElseWatch
    添加延时任务;该方法有两个入参:operation和watchkeys;前者是要添加的延时任务,后者是该任务要监听的事件类型(可以同时监听多个事件类型);可以在kafka.server.DelayedOperationKey所在scala文件中查看目前可选的watchkeys类型;一个watchkey对应一个延时任务队列;

  2. checkAndComplete
    检查各任务是否以满足提前触发的条件;该方法在某个事件发生之后,逻辑是遍历watchkey对应的任务队列,逐个判断是否满足了触发条件;

  3. cancelForKey
    取消watchkey下的所有延时任务;

三. 实现

下文主要关注"延时"的实现方式。

3.1 业务模型

时间轮延时组件的思路如下:

  1. 仍然使用java的DelayQueue存储延时任务;
    但为了降低延时任务入队的事件复杂度,Kafka并不直接将单个延时任务直接存储于DelayQueue,而是先将延时任务列表(TimerTaskList)加入DelayQueue,然后再向对应的列表中添加延时任务 (TimerTaskList也有超时时间的概念,等于其中最快超时的任务的超时时间)。这样任务的入队就由向优先级队列添加元素变为了向TimerTaskList中添加元素;前者时间复杂度为O(logn),后者时间复杂度为O(1)。

  2. 如何判断一个新的延时任务应该加到哪个TimerTaskList中?时间轮(TimingWheel)!
    delayQueue中存储了多个TimerTaskList,当拿到一个新的延时任务时,Kafka会使用TimingWheel来计算该延时任务应插入到哪个TimerTaskList。其实,TimingWheel的本质就是TimerTask和TimerTaskList之间的映射函数。

接下来通过一个具体的例子来说明这种映射逻辑:
Kafka源码分析(五) - Server端 - 基于时间轮的延时组件-LMLPHP
先关注上图中①号时间轮。圆环中每一个单元格表示一个TimerTaskList。单元格有其关联的时间跨度;下方的"1s x 12"表示时间轮上共12个单元格,每个单元格的时间跨度为1秒。有一个指针指向了"当前时间"所对应的单元格。顺时针方向为时间流动方向。

当收到一个延迟时间在0-1s的TimerTask时,会将其追加到①号时间轮的橙色单元格中。当收到一个延时时间在3-4s的TimerTask时,会将其追加到①号时间轮的黄色单元格中。以此类推。

现在有个问题:①号时间轮能表示的最大延迟时间是12秒,那如果收到了延迟13秒的任务该怎么办?这时该用到②号时间轮了,我们称②号为①号的"溢出时间轮"。②号时间轮的特点如下:

  1. 一个单元格所标识的时间跨度为①号的总长度,即12秒;
  2. 单元格数量与①号相同;

如此,延迟时间在12-24s的TimerTask会被追加到②号的紫色单元格,延迟时间在36-48s的TimerTask会被追加到②号的绿色单元格中。③号时间轮同理。

刚刚是按①->②->③的顺序来分析时间轮的逻辑,反过来也可以得到有用的结论:想象手里有一个"放大镜",其实③号时间轮的蓝色单元格"放大"后是②号时间轮;②号时间轮的蓝色单元格"放大"后是①号时间轮;蓝色单元格并不实际存储TimerTask。

3.2 数据结构

Kafka源码分析(五) - Server端 - 基于时间轮的延时组件-LMLPHP
DelayedOperationPurgatory有个Timer类型的timeoutTimer属性,用于维护延时任务。实际使用的是Timer的实现类:SystemTimer。该类用于维护延时任务的核心属性有两个:delayQueue和timingWheel。TimingWheel表示单个时间轮,接下来我们来看看其类图:
Kafka源码分析(五) - Server端 - 基于时间轮的延时组件-LMLPHP

各属性含义如下:

  1. buckets:TimerTaskList数组;其中一个元素对应一个时间轮单元格;

  2. tickMs:本时间轮一个单元格所表示的时间跨度(毫秒);

  3. intervel:本时间轮所能表示的时间总长度,其值为 b u c k e t s . l e n g t h ∗ t i c k M s buckets.length*tickMs buckets.lengthtickMs ;

  4. currentTime:当前时间;其是tickMs的整数倍,这样可以通过currentTime的值很方便地计算出"当前"对应的时间轮单元格;其初始值计算公式如下:
    currentTime = startMs - (startMs % tickMs );(其中startMs为时间轮创建时间)

  5. overflowWheel:本时间轮的"溢出时间轮";

3.3 算法

3.3.1 添加任务

添加任务的入口是 DelayedOperationPurgatory.tryCompleteElseWatch,其核心逻辑分如下两步:

def tryCompleteElseWatch(operation: T, watchKeys: Seq[Any]): Boolean = {
  // 1. 尝试立即指定指定的任务
  ...

  // 2. 加入到Timer
  timeoutTimer.add(operation)
  ...
}

SystemTimer.add直接调用了addTimerTaskEntry方法,后者逻辑如下:

private def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {
  // 尝试将任务加入时间轮
  if (!timingWheel.add(timerTaskEntry)) {
    // 若任务已过期且未被取消,直接提交执行
    if (!timerTaskEntry.cancelled)
      taskExecutor.submit(timerTaskEntry.timerTask)
  }
}

TimingWheel.add的逻辑也很清晰,分如下4种场景处理:

def add(timerTaskEntry: TimerTaskEntry): Boolean = {
  val expiration = timerTaskEntry.expirationMs

  if (timerTaskEntry.cancelled) {
    // 1.如果任务已取消
    ... ...
    false
  } else if (expiration < currentTime + tickMs) {
    // 2.若任务已过期, 即任务过期时间在"当前单元格"内
    ... ...
    false
  } else if (expiration < currentTime + interval) {
    // 3.虽然过期时间不在"当前单元格"内,但仍在当前时间轮之内
    // 将任务正常加入到当前时间轮对应的单元格内 
    ... ...
    true
  } else {
    // 4.超出了当前时间轮所能表示的范围: 将任务加入到"溢出时间轮"
    if (overflowWheel == null) 
      // 如果当前不存在上层时间轮, 那么生成一个
      addOverflowWheel()
    // 将任务加入到"溢出时间轮"
    overflowWheel.add(timerTaskEntry)
  }
}

3.3.2 尝试提前触发任务

入口是DelayedOperationPurgatory.checkAndComplete:

def checkAndComplete(key: Any): Int = {
  // 1. 根据事件key查找对应的任务队列(watchers就是一个队列的封装);
  val watchers = inReadLock(removeWatchersLock) { watchersForKey.get(key) }

  // 2. 遍历队列,尝试执行各任务;
  if(watchers == null)
    0
  else
    watchers.tryCompleteWatched()
}

接下来看Watchers.tryCompleteWatched方法的内容:

def tryCompleteWatched(): Int = {
  var completed = 0

  // 遍历队列
  val iter = operations.iterator()
  while (iter.hasNext) {
    // 逐个处理各任务
    val curr = iter.next()
    if (curr.isCompleted) {
      // 若任务已经处于完成态, 直接移除
      iter.remove()
    } else if (curr.maybeTryComplete()) {
      // 若任务在调用maybeTryComplete后被成功触发, 则移除
      iter.remove()
      completed += 1
    }
  }

  // 若队列已空, 移除当前队列
  if (operations.isEmpty)
    removeKeyIfEmpty(key, this)

  // 返回本次成功提前触发的任务数量
  completed
}

DelayedOperation.maybeTryComplete方法最终调用了DelayedOperation.tryComplete;

DelayedOperation的子类需要在后者中实现自己的"触发条件"检查逻辑;若满足了提前触发的条件,则调用forceComplete方法执行事件触发场景下的业务逻辑。

3.3.3 任务到期自动执行

DelayedOperationPurgatory中维护了一个expirationReaper线程,其职责就是循环调用kafka.utils.timer.SystemTimer#advanceClock来从事件轮中获取已超时的任务,并更新时间轮的"当前时间"指针。

四. 总结

才疏学浅,未能窥其十之一二,随时欢迎各位交流补充。若文章质量还算及格,可以点赞收藏加以鼓励,后续我继续更新。

另外也可以在目录中找到同系列的其他文章:
Kafka源码分析系列-目录(收藏关注不迷路)

感谢阅读。

05-13 05:10