在kafka中，有許多請(qǐng)求并不是立即返回，而且處理完一些異步操作或者等待某些條件達(dá)成后才返回，這些請(qǐng)求一般都會(huì)帶有timeout參數(shù)，表示如果timeout時(shí)間后服務(wù)端還不滿(mǎn)足返回的條件，就判定此次請(qǐng)求為超時(shí)，這時(shí)候kafka同樣要返回超時(shí)的響應(yīng)給客戶(hù)端，這樣客戶(hù)端才知道此次請(qǐng)求超時(shí)了。比如ack=-1的producer請(qǐng)求，就需要等待所有的isr備份完成了才可以返回給客戶(hù)端，或者到達(dá)timeout時(shí)間了返回超時(shí)響應(yīng)給客戶(hù)端。

上面的場(chǎng)景，可以用延遲任務(wù)來(lái)實(shí)現(xiàn)。也就是定義一個(gè)任務(wù)，在timeout時(shí)間后執(zhí)行，執(zhí)行的內(nèi)容一般就是先檢查返回條件是否滿(mǎn)足，滿(mǎn)足的話(huà)就返回客戶(hù)端需要的響應(yīng)，如果還是不滿(mǎn)足，就發(fā)送超時(shí)響應(yīng)給客戶(hù)端。

對(duì)于延遲操作，java自帶的實(shí)現(xiàn)有Timer和ScheduledThreadPoolExecutor。這兩個(gè)的底層數(shù)據(jù)結(jié)構(gòu)都是基于一個(gè)延遲隊(duì)列，在準(zhǔn)備執(zhí)行一個(gè)延遲任務(wù)時(shí)，將其插入到延遲隊(duì)列中。這些延遲隊(duì)列其實(shí)就是一個(gè)用最小堆實(shí)現(xiàn)的優(yōu)先級(jí)隊(duì)列，因此，插入一個(gè)任務(wù)的時(shí)間復(fù)雜度是O(logN),取出一個(gè)任務(wù)執(zhí)行后調(diào)整堆的時(shí)間也是O(logN)。

如果要執(zhí)行的延遲任務(wù)不多，O(logN)的速度已經(jīng)夠快了。但是對(duì)于kafka這樣一個(gè)高吞吐量的系統(tǒng)來(lái)說(shuō)，O(logN)的速度還不夠，為了追求更快的速度，kafka的設(shè)計(jì)者使用了Timing Wheel的數(shù)據(jù)結(jié)構(gòu)，讓任務(wù)的插入時(shí)間復(fù)雜度達(dá)到了O(1)。

Timing Wheel

上面是時(shí)間輪的一個(gè)結(jié)構(gòu)圖，該時(shí)間輪有8個(gè)槽，當(dāng)前時(shí)間指向0號(hào)槽。

我們?cè)倏匆幌翶afka里面TimingWheel的數(shù)據(jù)結(jié)構(gòu)

private[timer] class TimingWheel(tickMs: Long, wheelSize: Int, startMs: Long, taskCounter: AtomicInteger, queue: DelayQueue[TimerTaskList]) {
  private[this] val interval = tickMs * wheelSize  private[this] val buckets = Array.tabulate[TimerTaskList](wheelSize) { _ => new TimerTaskList(taskCounter) }
  private[this] var currentTime = startMs - (startMs % tickMs) // rounding down to multiple of tickMs}

tickMs：表示一個(gè)槽所代表的時(shí)間范圍，kafka的默認(rèn)值的1ms

wheelSize：表示該時(shí)間輪有多少個(gè)槽，kafka的默認(rèn)值是20

startMs：表示該時(shí)間輪的開(kāi)始時(shí)間

taskCounter：表示該時(shí)間輪的任務(wù)總數(shù)

queue：是一個(gè)TimerTaskList的延遲隊(duì)列。每個(gè)槽都有它一個(gè)對(duì)應(yīng)的TimerTaskList，TimerTaskList是一個(gè)雙向鏈表，有一個(gè)expireTime的值，這些TimerTaskList都被加到這個(gè)延遲隊(duì)列中，expireTime最小的槽會(huì)排在隊(duì)列的最前面。

interval：時(shí)間輪所能表示的時(shí)間跨度，也就是tickMs*wheelSize

buckets：表示TimerTaskList的數(shù)組，即各個(gè)槽。

currentTime：表示當(dāng)前時(shí)間，也就是時(shí)間輪指針指向的時(shí)間

運(yùn)行原理

當(dāng)新增一個(gè)延遲任務(wù)時(shí)，通過(guò)buckets[expiration / tickMs % wheelSize]先計(jì)算出它應(yīng)該屬于哪個(gè)槽。比如延遲任務(wù)的delayMs=2ms，當(dāng)前時(shí)間currentTime是0ms，則expiration=delayMs+startMs=2ms，通過(guò)前面的公式算出它應(yīng)該落于2號(hào)槽。并把任務(wù)封裝成TimerTaskEntry然后加入到TimerTaskList鏈表中。

之后，kafka會(huì)啟動(dòng)一個(gè)線(xiàn)程，去推動(dòng)時(shí)間輪的指針轉(zhuǎn)動(dòng)。其實(shí)現(xiàn)原理其實(shí)就是通過(guò)queue.poll()取出放在最前面的槽的TimerTaskList。由于queue是一個(gè)延遲隊(duì)列，如果隊(duì)列中的expireTime沒(méi)有到達(dá)，該操作會(huì)阻塞住，直到expireTime到達(dá)。如果通過(guò)queue.poll()取到了TimerTaskList，說(shuō)明該槽里面的任務(wù)時(shí)間都已經(jīng)到達(dá)。這時(shí)候就可以遍歷該TimerTaskList中的任務(wù)，然后執(zhí)行對(duì)應(yīng)的操作了。

針對(duì)上面的例子，就2號(hào)槽有任務(wù)，所以當(dāng)取出2號(hào)槽的TimerTaskList后，會(huì)先將currentTime = timeMs - (timeMs % tickMs)，其中timeMs也就是該TimerTaskList的expireTime，也就是2Ms。所以，這時(shí)currentTime=2ms，也就是時(shí)間輪指針指向2Ms。

時(shí)間溢出處理

在kafka的默認(rèn)實(shí)現(xiàn)中，tickMs=1Ms，wheelSize=20，這就表示該時(shí)間輪所能表示的延遲時(shí)間范圍是0~20Ms，那如果延遲時(shí)間超過(guò)20Ms要如何處理呢？Kafka對(duì)時(shí)間輪做了一層改進(jìn)，使時(shí)間輪變成層級(jí)的時(shí)間輪。

一開(kāi)始，第一層的時(shí)間輪所能表示時(shí)間范圍是0~20Ms之間，假設(shè)現(xiàn)在出現(xiàn)一個(gè)任務(wù)的延遲時(shí)間是200Ms，那么kafka會(huì)再創(chuàng)建一層時(shí)間輪，我們稱(chēng)之為第二層時(shí)間輪。

第二層時(shí)間輪的創(chuàng)建代碼如下

overflowWheel = new TimingWheel(          tickMs = interval,          wheelSize = wheelSize,          startMs = currentTime,          taskCounter = taskCounter,          queue)

也就是第二層時(shí)間輪每一個(gè)槽所能表示的時(shí)間是第一層時(shí)間輪所能表示的時(shí)間范圍，也就是20Ms。槽的數(shù)量還是一樣，其他的屬性也是繼承自第一層時(shí)間輪。這時(shí)第二層時(shí)間輪所能表示的時(shí)間范圍就是0~400Ms了。

之后通過(guò)buckets[expiration / tickMs % wheelSize]算出延遲時(shí)間為200Ms的任務(wù)應(yīng)該位于第二層時(shí)間輪的10號(hào)槽位。

同理，如果第二層時(shí)間輪的時(shí)間范圍還容納不了新的延遲任務(wù)，就會(huì)創(chuàng)建第三層、第四層...

值得注意的是，只有當(dāng)前時(shí)間輪無(wú)法容納目標(biāo)延遲任務(wù)所能表示的時(shí)間時(shí)，才需要?jiǎng)?chuàng)建更高一級(jí)的時(shí)間輪，或者說(shuō)把該任務(wù)加到更高一級(jí)的時(shí)間輪中(如果該時(shí)間輪已創(chuàng)建)。

一些細(xì)節(jié)

1. 當(dāng)時(shí)間輪的指針指向1號(hào)槽時(shí)，即currentTime=1Ms，說(shuō)明0號(hào)槽的任務(wù)都已經(jīng)到期了，這時(shí)0號(hào)槽就會(huì)被拿出來(lái)復(fù)用，可以容納20~21Ms延遲時(shí)間的任務(wù)。也就是說(shuō)，如果currentTime=0Ms時(shí)進(jìn)來(lái)一個(gè)21Ms的延遲任務(wù)，就需要?jiǎng)?chuàng)建更高一級(jí)的時(shí)間輪，但是如果currentTime=1Ms時(shí)進(jìn)來(lái)一個(gè)21Ms的延遲任務(wù)，就可以直接把它放到0號(hào)槽中，當(dāng)currentTime=21時(shí)，指針又指向0號(hào)槽

2. 細(xì)心的同學(xué)可能發(fā)現(xiàn)，第一層的0號(hào)槽所能表示的任務(wù)延遲時(shí)間范圍是0_{1Ms，對(duì)應(yīng)的TimerTaskList的expireTime是0Ms。第二層的0號(hào)槽鎖能表示的任務(wù)延遲時(shí)間范圍是0}20Ms，對(duì)應(yīng)的TimerTaskList的expireTime也是0Ms。他們的TimerTaskList又都是放在一個(gè)延遲隊(duì)列中。這時(shí)候執(zhí)行queue.poll()會(huì)把這兩個(gè)TimerTaskList都取出來(lái)，然后遍歷鏈表的時(shí)候還會(huì)判斷該任務(wù)是否達(dá)到執(zhí)行時(shí)間了，如果沒(méi)有的話(huà)，這些任務(wù)還會(huì)被塞回時(shí)間輪中。這時(shí)由于第一層指針的轉(zhuǎn)動(dòng)，原先處于第二層時(shí)間輪中的任務(wù)可能會(huì)重新落到第一層時(shí)間輪上面。

源碼解析

添加新的延遲任務(wù)

//SystemTimer.scala  private def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {    if (!timingWheel.add(timerTaskEntry)) {      // Already expired or cancelled      if (!timerTaskEntry.cancelled)        taskExecutor.submit(timerTaskEntry.timerTask)    }  }

往時(shí)間輪添加新的任務(wù)

//TimingWheeldef add(timerTaskEntry: TimerTaskEntry): Boolean = {    //獲取任務(wù)的延遲時(shí)間    val expiration = timerTaskEntry.expirationMs    //先判斷任務(wù)是否已經(jīng)完成    if (timerTaskEntry.cancelled) {      false      //如果任務(wù)已經(jīng)到期    } else if (expiration < currentTime + tickMs) {      false      //判斷當(dāng)前時(shí)間輪所能表示的時(shí)間范圍是否可以容納該任務(wù)    } else if (expiration < currentTime + interval) {      // 根據(jù)任務(wù)的延遲時(shí)間算出應(yīng)該位于哪個(gè)槽      val virtualId = expiration / tickMs      val bucket = buckets((virtualId % wheelSize.toLong).toInt)      bucket.add(timerTaskEntry)
      // 設(shè)置TimerTaskList的expireTime      if (bucket.setExpiration(virtualId * tickMs)) {        //把TimerTaskList加入到延遲隊(duì)列        queue.offer(bucket)      }      true    } else {      //如果時(shí)間超出當(dāng)前所能表示的最大范圍，則創(chuàng)建新的時(shí)間輪,并把任務(wù)添加到那個(gè)時(shí)間輪上面      if (overflowWheel == null) addOverflowWheel()      overflowWheel.add(timerTaskEntry)    }  }  private[this] def addOverflowWheel(): Unit = {    synchronized {      if (overflowWheel == null) {        overflowWheel = new TimingWheel(          tickMs = interval,          wheelSize = wheelSize,          startMs = currentTime,          taskCounter = taskCounter,          queue        )      }    }  }

從上面的代碼可以看出，對(duì)于當(dāng)前時(shí)間輪是否可以容納目標(biāo)任務(wù)，是通過(guò)expiration < currentTime + interval來(lái)計(jì)算的，也就是根據(jù)時(shí)間輪的指針往后推interval時(shí)間就是時(shí)間輪所能表示的時(shí)間范圍。

時(shí)間輪指針的推進(jìn)

 //SystemTimer.scala def advanceClock(timeoutMs: Long): Boolean = {      //從延遲隊(duì)列中取出最近的一個(gè)槽，如果槽的expireTime沒(méi)到，此操作會(huì)阻塞timeoutMs    var bucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS)    if (bucket != null) {      writeLock.lock()      try {        while (bucket != null) {            //推進(jìn)時(shí)間輪的指針          timingWheel.advanceClock(bucket.getExpiration())            //把TimerTaskList的任務(wù)都取出來(lái)重新add一遍，add的時(shí)候會(huì)檢查任務(wù)是否已經(jīng)到期          bucket.flush(reinsert)          bucket = delayQueue.poll()        }      } finally {        writeLock.unlock()      }      true    } else {      false    }  }//TimingWheeldef advanceClock(timeMs: Long): Unit = {    if (timeMs >= currentTime + tickMs) {        //推進(jìn)時(shí)間輪的指針      currentTime = timeMs - (timeMs % tickMs)
      // 推進(jìn)上層時(shí)間輪的指針      if (overflowWheel != null) overflowWheel.advanceClock(currentTime)    }  }

總結(jié)

相比于常用的DelayQueue的時(shí)間復(fù)雜度O(logN)，TimingWheel的數(shù)據(jù)結(jié)構(gòu)在插入任務(wù)時(shí)只要O(1),獲取到達(dá)任務(wù)的時(shí)間復(fù)雜度也遠(yuǎn)低于O(logN)。另外，kafka的TimingWheel在插入任務(wù)之前還會(huì)先檢查任務(wù)是否完成，對(duì)于那些在任務(wù)超時(shí)直接就完成指定操作的場(chǎng)景，TimingWheel的表現(xiàn)更加優(yōu)秀。

- END -