• 在kafka中有很多操作需要延遲等待, 比如客戶端發(fā)送數(shù)據(jù)到達(dá)leader后, 根據(jù)設(shè)置ack方式不同,需要等待其replicas返回ack, 那這個(gè)ack就需要延遲等待;對于一個(gè)拉取操作, 需要延遲等待期望拉取的字節(jié)數(shù)準(zhǔn)備好;
• 有延遲操作, 那必然會存在操作的超時(shí)處理. 還記得我們上一篇Kafka中的時(shí)間輪中對Timer的分析吧, 這里的延遲操作需要使用它來實(shí)現(xiàn);

DelayedOperation

• 所在文件: core/src/main/scala/kafka/server/DelayedOperation.scala
• 這是個(gè)抽象類, 所有具體的延遲任務(wù)都需要繼承這個(gè)類;
• 同時(shí)每個(gè)延遲任務(wù)必然存在操作的超時(shí), 那么其超時(shí)操作是通過將對象放到Kafka中的時(shí)間輪中的Timer中處理, 因此這個(gè)類又繼承自TimerTask;
• private val completed = new AtomicBoolean(false): 原子變量, 標(biāo)識當(dāng)前operation是否已完成;
• def forceComplete(): Boolean: 強(qiáng)制完成操作;

if (completed.compareAndSet(false, true)) {
      // cancel the timeout timer
      cancel()
      onComplete()
      true
    } else {
      false
    }

分兩種情況:

1. 當(dāng)前操作已經(jīng)完成,則不再需要強(qiáng)制完成,返回false;
2. 當(dāng)前操作未完成, 則首先在Timer中取消這個(gè)定時(shí)任務(wù), 然后回調(diào)onComplete

• override def run(): Unit: 實(shí)現(xiàn)的是TimerTask的方法, 當(dāng)超時(shí)時(shí)會執(zhí)行此操作:

if (forceComplete())
      onExpiration()

里面的操作比較簡單, 調(diào)用forceComplete, 如果成功,表明是真的超時(shí)了,回調(diào)onExpiration;

• 需要由子類實(shí)現(xiàn)的方法:

1. def onExpiration(): Unit: 超時(shí)后的回調(diào)處理;
2. def onComplete(): Unit: 操作完成后的回調(diào)處理;
3. def tryComplete(): Boolean: 在放入到Timer前, 先嘗試著執(zhí)行一下這個(gè)操作, 看是否可以完成, 如果可以就不用放到Timer里了, 這是為了確保任務(wù)都盡快完成作的一個(gè)優(yōu)化;

Watchers

• 所在文件: core/src/main/scala/kafka/server/DelayedOperation.scala
• 對于一個(gè)延遲任務(wù), 一般會有兩個(gè)操作加持在身:

1. 上面說的作為超時(shí)任務(wù)放在Timer中;
2. 與某些事件關(guān)聯(lián)在一起, 可以關(guān)聯(lián)多個(gè)事件, 當(dāng)這些事件中的某一個(gè)發(fā)生時(shí), 這個(gè)任務(wù)即可認(rèn)為是完成;這個(gè)就是 Watchers類要完成的工作;

• class Watchers(val key: Any): 構(gòu)造時(shí)需要一個(gè)參數(shù)key, 你可以理解成是一個(gè)事件;
• private[this] val operations = new LinkedList[T](): 用于存放和這個(gè)key關(guān)聯(lián)的所有操作,一個(gè)key可以關(guān)聯(lián)多個(gè)操作, 同時(shí)一個(gè)操作也可以被多個(gè)key關(guān)聯(lián)(即位于多個(gè)Watchers對象中)
• def purgeCompleted(): Int: 刪除鏈表中所有已經(jīng)完成的操作

      var purged = 0
      operations synchronized {
        val iter = operations.iterator()
        while (iter.hasNext) {
          val curr = iter.next()
          if (curr.isCompleted) {
            iter.remove()
            purged += 1
          }
        }
      }
      if (operations.size == 0)
        removeKeyIfEmpty(key, this)

      purged
    }

• def tryCompleteWatched(): Int:

     var completed = 0
      operations synchronized {
        val iter = operations.iterator()
        while (iter.hasNext) {
          val curr = iter.next()
          if (curr.isCompleted) {
            // another thread has completed this operation, just remove it
            iter.remove()
          } else if (curr synchronized curr.tryComplete()) {
            completed += 1
            iter.remove()
          }
        }
      }

      if (operations.size == 0)
        removeKeyIfEmpty(key, this)

      completed

掃描整個(gè)鏈表:

1. 如果任務(wù)已完成,則移除;
2. 如果任務(wù)未完成, 調(diào)用tryComplete嘗試立即完成, 如果可以完成, 則移除;

• 添加任務(wù):

def watch(t: T) {
      operations synchronized operations.add(t)
    }

DelayedOperationPurgatory

• 所在文件: core/src/main/scala/kafka/server/DelayedOperation.scala
• 終于要揭開我們的謎之煉獄啦, 源碼里的注釋如下:

A helper purgatory class for bookkeeping delayed operations with a timeout, and expiring timed out operations

實(shí)際上就是用來通過Timer和Watchers來管理一批延遲任務(wù);

• private[this] val timeoutTimer = new Timer(executor): 用來處理加入的作務(wù)的超時(shí)行為;
• private val expirationReaper = new ExpiredOperationReaper():

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

    override def doWork() {
      timeoutTimer.advanceClock(200L)

      if (estimatedTotalOperations.get - delayed > purgeInterval) {
        estimatedTotalOperations.getAndSet(delayed)
        debug("Begin purging watch lists")
        val purged = allWatchers.map(_.purgeCompleted()).sum
        debug("Purged %d elements from watch lists.".format(purged))
      }
    }
  }

1. timeoutTimer.advanceClock(200L): 驅(qū)動Timer向前走, pop出超時(shí)的延遲任務(wù);
2. val purged = allWatchers.map(_.purgeCompleted()).sum: 利用閾值(purgeInterval)來周期性地從Watchers中清理掉已經(jīng)完成的任務(wù);

• def tryCompleteElseWatch(operation: T, watchKeys: Seq[Any]): Boolean: 將operation和一系列的事件(key)關(guān)聯(lián)起來, 然后調(diào)用tryComplete嘗試立即完成該操作,如果不能完成,加入到Timer中;

• def checkAndComplete(key: Any): Int: 按key找到相應(yīng)的Watchers對象, 然后調(diào)用其tryCompleteWatched(), 說明上面用;

簡圖:

DelayedOperation.png

基本上就是這些了

Kafka源碼分析-匯總