Kafka的延遲操作是一個相對獨立的元件，他的主要功能是管理延遲操作，底層依賴於Kafka提供的時間輪實現。JDK本身提供的java.util.Timer也可以實現定時任務，但是如果系統請求量巨大，效能要求很高，他們底層所依賴的資料結構存取操作複雜度都是O(nlog(n))

為了將時間複雜度降為o(1)，一般會使用其他方式的定時任務元件，比如zookeeper的時間桶方式處理session過期，netty也使用Hash

WheelTimer這種時間輪的實現。

Kafka時間輪的實現是TimingWheel，他是一個儲存定時任務的環形佇列（桶），底層使用陣列實現，陣列中每一個元素可以存放一個TimerTaskList物件

TimerTaskList是環形雙向連結串列，在其中連結串列項TimeTaskEntry封裝了真正的定時任務TimerTask。TimerTaskList使用expiration欄位記錄了整個TimerTaskList的超時時間。TimeTaskEntry中的expirationMs欄位記錄了超時時間戳，timerTask欄位指向了對應的TimerTask任務.

TimerTask中的delayMs記錄了任務的延遲時間，timerTaskEntry記錄了TimerTaskEntry物件

TimingWheel提供了分層的概念，因為年時間跨度比較大，數量很大，單層的時間輪會造成任務的round很大，單個格子連結串列很長。一般情況，第一層時間跨度是最小的，第二層時間跨度比較大。

如上圖所示：假設編號為0的時間格或者桶儲存著到期時間為t,每一個tick的持續時間（tickDuration）為20ms，在這個格子裡只能儲存著到期時間為[t~t+20]ms的任務，任務到底放在哪一個時間格或者桶裡面根據不同的場景可以有不同的演算法，假設時間輪的時間格有n個，到期時間為m(ms),那麼計算公式m%n = 所在的時間格或者桶，比如n=10,m=34ms,那麼他所在桶或者時間格是4

當任務到期時間超出了當前時間所表示的時間範圍時，就會嘗試加到上一層時間輪，如下圖所示：

其中第一層時間輪每個時間格是1ms,整個時間輪跨度是20ms，指標當前時間表示的時間是currentTime，則該時間輪跨度為currentTime

~currentTime+20，只有時間在這段範圍內任務才能新增到該層時間輪等待到期。到期時間超出[currentTime~currentTime+20]這個時間範圍的任務會嘗試新增到上級時間輪中，通過逐層向上級嘗試最終找到合適的時間輪層級

整個時間輪表示的時間跨度是不變的，隨著指標的不斷後移，當前時間輪能處理的時間段也在不斷後移，新來的TimerTaskEntry會複用原來的已經到期的TimerTaskList，如下圖所示，第一層時間輪跨度始終為20ms,指標表示的時間在不段後移。當指標指向0是時間格的時候，假設currentTime = 100，指向第三個時間格，此時指標表示的時間為當前時間104ms,整個時間輪表示的時間段是[104~ 124]，但是該時間輪的時間跨度依然是20ms。此時間輪中編號為2的時間格表示的時間不再是102~103,而是123~124

假設現在有一個任務在445ms後執行，預設情況下，各個層級的時間輪的時間格個數為20，第一層時間輪每一個時間格跨度為1ms,整個時間輪跨度為20ms,跨度不夠。第二層時間輪每一個時間格跨度為20ms,整個時間輪跨度為400ms,跨度依然不夠，第三層時間輪每一個時間格跨度為400ms,整個時間輪跨度為8000ms，現在跨度夠了，此任務就放在第三層時間輪的第一個時間格對應的TimerTaskList，等待被執行，此TimerTaskList到期時間是400ms,隨著時間的流逝，當此TimerTaskList到期時，距離該任務到期時間還有45ms，不能執行該任務，我們將重新提交到時間輪，此時第一層時間輪跨度依然不夠，不能執行任務，第二層時間輪時間格跨度為20，整個世間輪跨度為400，跨度足夠，放在第三個時間格等待執行，如此往復幾次，高層時間輪最終會慢慢移動到低層時間輪上，最終任務到期執行。

一 重要屬性

buckets : Array.tabulate[TimerTaskList] 型別，其每一個項都對應時間輪中一個時間格，用於儲存TimerTaskList的陣列

tickMs：Long 當前時間輪中一個時間格表示的時間跨度

wheelSize: Int 當前時間輪的大小也就是總的時間格數量

taskCounter：AtomicInteger 各層級時間輪中任務的總數

startMs：Long 當期時間輪的建立時間

queue:DelayQueue[TimerTaskList] 整個層級的時間輪公用一個任務佇列，其元素型別是TimerTaskList

currentTime：時間輪的指標，將整個時間輪劃分為到期部分和未到期部分。在初始化的時候，currentTime被修剪成tickMs的倍數startMs - (startMs % tickMs)

interval：Long 當前時間輪的時間跨度即tickMs * wheelSize,當前時間輪只能處理時間範圍在currentTime~currentTime+tickMs*WheelSize之間的定時任務，超過這個範圍則需要新增任務到上層時間輪

overflowWheel: TimingWheel 上層時間輪的引用

二 核心方法

2.1 addOverflowWheel 主要用於建立上層時間輪

private[this] def addOverflowWheel(): Unit = {
  synchronized {
    if (overflowWheel == null) {
      overflowWheel = new TimingWheel(
        tickMs = interval, // 上層時間輪的時間格跨度等於下一層時間輪的跨度wheelSize = wheelSize,// 大小不變startMs = currentTime,// 初始化當前時間輪的建立時間taskCounter= taskCounter,//時間輪中任務的總數queue
      )
    }
  }
}

2.2 add 向時間輪中新增定時任務，同時也會檢測新增的任務是否已經到期

def add(timerTaskEntry: TimerTaskEntry): Boolean = {
  // 獲取定時任務的超時時間戳val expiration = timerTaskEntry.expirationMs
  // 如果任務已經被取消if (timerTaskEntry.cancelled) {
    false // 返回新增失敗} else if (expiration < currentTime + tickMs) { // 如果時間指標現在指向的時間+時間格跨度 > 需要新增的定時任務的超時時間戳表示已經到期
    // 舉個例子：currentTime=102，時間格跨度為10ms,那麼假設新增的任務超時時間戳為105 < 102+10
    false // 返回新增失敗} else if (expiration < currentTime + interval) {// 如果時間指標現在指向的時間+當前時間輪跨度 > 需要新增的定時任務的超時時間戳表示沒有到期
    // 然後把當前任務放進迴圈數組裡面
    // 得到任務到期時間戳/時間格跨度的餘數val virtualId = expiration / tickMs
    // 獲取放在哪一個桶裡 (到期時間戳/時間格跨度)%時間輪跨度val bucket = buckets((virtualId % wheelSize.toLong).toInt)
    bucket.add(timerTaskEntry)

    // 設定時間格的到期時間if (bucket.setExpiration(virtualId * tickMs)) {
      /*
       * 整個時間輪表示的跨度是不變的，隨著指標的後移，當前時間輪能夠處理的時間段也在不段後移，新的TimerTaskEntry會複用原來的已經清理過的
       * TimerTaskList，此時你需要重新設定TimerTaskList的到期時間，並將桶重新入隊
       */
      queue.offer(bucket)
    }
    true
  } else { // 如果超出了時間的跨度範圍，則將其新增到上層時間輪來處理if (overflowWheel == null) addOverflowWheel()
    overflowWheel.add(timerTaskEntry)
  }
}

2.3 advanceClock 嘗試推進當前時間輪的指標，同時也會嘗試推進上層時間輪的指標，隨著當前時間輪的不斷推進，上層時間輪指標早晚會被推進成功

def advanceClock(timeMs: Long): Unit = {
  // 嘗試移動指標currentTime
  if (timeMs >= currentTime + tickMs) {
    currentTime = timeMs - (timeMs % tickMs)

    // 嘗試禿頂上層時間輪指標currentTime
    if (overflowWheel != null) overflowWheel.advanceClock(currentTime)
  }
}