深入理解Kafka服务端之延时请求的处理

trait Timer {


   将给定的定时任务插入到时间轮上，等待后续延迟执行
  def add(timerTask: TimerTask): Unit


   向前推进时钟，执行已达过期时间的延迟任务
  def advanceClock(timeoutMs: Long): Boolean


   获取时间轮上总的定时任务数
  def size: Int


   关闭定时器
  def shutdown(): Unit
}

@threadsafe
class SystemTimer(executorName: String,//Purgatory 的名字。Kafka 中存在不同的 Purgatory
                  tickMs: Long = 1,//时间轮中时间格的跨度
                  wheelSize: Int = 20,//时间轮上时间格的数量
                  startMs: Long = Time.SYSTEM.hiResClockMs//该 SystemTimer 定时器启动时间，单位是毫秒。
                 ) extends Timer {


  // 单线程的线程池用于异步执行定时任务
  private[this] val taskExecutor = Executors.newFixedThreadPool(1, new ThreadFactory() {
    def newThread(runnable: Runnable): Thread =
      KafkaThread.nonDaemon("executor-"+executorName, runnable)
  })
  // 延迟队列保存所有Bucket，即所有TimerTaskList对象
  private[this] val delayQueue = new DelayQueue[TimerTaskList]()
  // 总定时任务数
  private[this] val taskCounter = new AtomicInteger(0)


  //时间轮对象
  private[this] val timingWheel = new TimingWheel(
    tickMs = tickMs,
    wheelSize = wheelSize,
    startMs = startMs,
    taskCounter = taskCounter,
    delayQueue
  )


  // 维护线程安全的读写锁
  private[this] val readWriteLock = new ReentrantReadWriteLock()
  private[this] val readLock = readWriteLock.readLock()
  private[this] val writeLock = readWriteLock.writeLock()
  ...
}

def add(timerTask: TimerTask): Unit = {
  // 获取读锁。在没有线程持有写锁的前提下，
  // 多个线程能够同时向时间轮添加定时任务
  readLock.lock()
  try {
    // 调用addTimerTaskEntry执行插入逻辑
    addTimerTaskEntry(new TimerTaskEntry(timerTask, timerTask.delayMs + Time.SYSTEM.hiResClockMs))
  } finally {
    //释放读锁
    readLock.unlock()
  }
}

private def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {


  if (!timingWheel.add(timerTaskEntry)) {


    // 定时任务未取消，说明定时任务已过期，就需要添加到执行任务的线程池等待执行
    // 否则timingWheel.add方法应该返回True
    if (!timerTaskEntry.cancelled)
      taskExecutor.submit(timerTaskEntry.timerTask)
  }
}

def advanceClock(timeoutMs: Long): Boolean = {


  //获取delayQueue中下一个已过期的Bucket，只有已经过期的任务才会执行，否则还是保存在时间轮中
  var bucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS)


  if (bucket != null) {
    //获取写锁
    // 一旦有线程持有写锁，其他任何线程执行add或advanceClock方法时会阻塞
    writeLock.lock()
    try {
      while (bucket != null) {
        // 推动时间轮向前"滚动"到Bucket的过期时间点
        timingWheel.advanceClock(bucket.getExpiration())
        // 将该Bucket下的所有定时任务重写回到时间轮，这是因为部分任务会进行降级
        bucket.flush(reinsert)
        // 读取下一个Bucket对象
        bucket = delayQueue.poll()
      }
    } finally {
      // 释放写锁
      writeLock.unlock()
    }
    true
  } else {
    false
  }
}

abstract class DelayedOperation(override val delayMs: Long,//超时时间，一般是客户端请求的超时时间，即request.timeout.ms
    lockOpt: Option[Lock] = None) extends TimerTask with Logging {
  // 标识该延迟操作是否已经完成
  private val completed = new AtomicBoolean(false)
  // 防止多个线程同时检查操作是否可完成时发生锁竞争导致操作最终超时
  private val tryCompletePending = new AtomicBoolean(false)


  private[server] val lock: Lock = lockOpt.getOrElse(new ReentrantLock)
  ...
}

private[server] def maybeTryComplete(): Boolean = {
  //标记是否需要重试
  var retry = false
  //标记延时操作是否已经完成
  var done = false
  do {
    //尝试获取锁对象
    if (lock.tryLock()) {
      try {
        tryCompletePending.set(false)
        //尝试完成延迟操作
        done = tryComplete()
      } finally {
        lock.unlock()
      }
      // 运行到这里的线程持有锁，其他线程只能运行else分支的代码
      // 如果其他线程将maybeTryComplete设置为true，那么retry=true
      // 这就相当于其他线程给了本线程重试的机会
      retry = tryCompletePending.get()
    } else {


      // 运行到这里的线程没有拿到锁
      // 设置tryCompletePending=true给持有锁的线程一个重试的机会
      retry = !tryCompletePending.getAndSet(true)
    }
  } while (!isCompleted && retry)
  done
}

final class DelayedOperationPurgatory[T <: DelayedOperation](purgatoryName: String,//Purgatory名称
                                                             timeoutTimer: Timer,//定时器，里面封装了分层时间轮
                                                             brokerId: Int = 0,//broker编号
                                                             purgeInterval: Int = 1000,//控制删除线程移除 Bucket 中的过期延迟请求的频率，默认为1秒
                                                             reaperEnabled: Boolean = true,//是否启动删除线程，默认为true
                                                             timerEnabled: Boolean = true)//是否启用分层时间轮，默认为true
                                                              extends Logging with KafkaMetricsGroup {
  private val watcherLists = Array.fill[WatcherList](DelayedOperationPurgatory.Shards)(new WatcherList)  


  //管理器中预估的总的延时请求数
  private[this] val estimatedTotalOperations = new AtomicInteger(0) 


  //过期操作收割机线程
  private val expirationReaper = new ExpiredOperationReaper()
  ...
}

 private class WatcherList {
  // 定义一组按照Key分组的Watchers对象，Key 可以是任何类型，而 Value 就是 Key 对应类型的一组 Watchers 对象。
  val watchersByKey = new Pool[Any, Watchers](Some((key: Any) => new Watchers(key)))


  val watchersLock = new ReentrantLock()
  // 返回所有Watchers对象
  def allWatchers = {
    watchersByKey.values
  }
}

private class Watchers(val key: Any) {
  //延迟请求的链表
  private[this] val operations = new ConcurrentLinkedQueue[T]()
  ...
}

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()) {
      iter.remove()
      completed += 1
    }
  }


  if (operations.isEmpty)
    removeKeyIfEmpty(key, this)
  //返回完成的延时请求数
  completed
}

private class ExpiredOperationReaper extends ShutdownableThread(
  "ExpirationReaper-%d-%s".format(brokerId, purgatoryName),
  false) {


  override def doWork() {
    advanceClock(200L)
  }
}

def tryCompleteElseWatch(operation: T, watchKeys: Seq[Any]): Boolean = {
  assert(watchKeys.nonEmpty, "The watch key list can't be empty")


  //尝试完成给定的操作
  var isCompletedByMe = operation.tryComplete()
  // 如果该延迟请求是由本线程完成的，直接返回true即可
  if (isCompletedByMe)
    return true


  var watchCreated = false
  // 遍历所有要监控的Key
  for(key <- watchKeys) {
    // 再次查看请求的完成状态，如果已经完成，就说明是被其他线程完成的，返回false
    if (operation.isCompleted)
      return false
    // 否则，将该operation加入到Key所在的WatcherList
    watchForOperation(key, operation)


    // 设置watchCreated标记，表明该任务已经被加入到WatcherList
    if (!watchCreated) {
      watchCreated = true
      // 更新Purgatory中总请求数
      estimatedTotalOperations.incrementAndGet()
    }
  }
  // 再次尝试完成该延迟请求
  isCompletedByMe = operation.maybeTryComplete()
  if (isCompletedByMe)
    return true


  // 如果依然不能完成此请求，将其加入到时间轮
  if (!operation.isCompleted) {
    if (timerEnabled)
      timeoutTimer.add(operation)
    //再次判断，如果该请求已完成，则取消时间轮中的定时任务
    if (operation.isCompleted) {
      //取消定时任务
      operation.cancel()
    }
  }


  false
}

def checkAndComplete(key: Any): Int = {
  // 获取给定Key对应的WatcherList
  val wl = watcherList(key)


  // 获取WatcherList中Key对应的Watchers对象实例，里面存储了需要完成的多个延时请求任务
  val watchers = inLock(wl.watchersLock) { wl.watchersByKey.get(key) }


  // 尝试完成满足完成条件的延时请求并返回成功完成的请求数
  if(watchers == null)
    0
  else
    //重要步骤：调用 Watchers 的 tryCompleteWatched 方法，尝试完成那些已满足完成条件的延迟请求
    watchers.tryCompleteWatched()
}

//是否需要等待其它副本完成写入，通过delayedProduceRequestRequired方法来判断写入是否成功，如果为true，表示需要等待其它副本同步完成消息写入
if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
  //创建延时生产任务对象
  val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
  //创建DelayedProduce延时请求对象
  val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)


  val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toSeq


  //再一次尝试完成该延时请求
  //如果暂时无法完成，则将对象放入到相应的Purgatory中等待后续处理
  delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)


} else {
  //可以立即返回
  val produceResponseStatus = produceStatus.mapValues(status => status.responseStatus)
  //调用该回调逻辑然后返回
  responseCallback(produceResponseStatus)
}

override def tryComplete(): Boolean = {


  //遍历请求中的分区，检查写到这些分区的数据是否已经同步到了所有Follower副本上
  produceMetadata.produceStatus.foreach { case (topicPartition, status) =>
    trace(s"Checking produce satisfaction for $topicPartition, current status $status")


    if (status.acksPending) {
      val (hasEnough, error) = replicaManager.getPartition(topicPartition) match {
        case Some(partition) =>
          if (partition eq ReplicaManager.OfflinePartition)
            (false, Errors.KAFKA_STORAGE_ERROR)
          else
            partition.checkEnoughReplicasReachOffset(status.requiredOffset)
        case None =>
     
          (false, Errors.UNKNOWN_TOPIC_OR_PARTITION)
      }


      if (error != Errors.NONE || hasEnough) {
        status.acksPending = false
        status.responseStatus.error = error
      }
    }
  }


  if (!produceMetadata.produceStatus.values.exists(_.acksPending))
    //强制完成延时请求，内部会首先会取消时间轮上的任务，然后调用 onComplete 方法
    forceComplete()
  else
    false
}

def forceComplete(): Boolean = {
  if (completed.compareAndSet(false, true)) {
    //取消时间轮上的定时任务
    cancel()
    //完成延时操作
    onComplete()
    true
  } else {
    false
  }
}

override def onComplete() {
  val responseStatus = produceMetadata.produceStatus.mapValues(status => status.responseStatus)
    //执行回调逻辑
    responseCallback(responseStatus)
}