如何解析Kafka中的時間輪問題

這期內容當中小編將會給大家帶來有關如何解析Kafka中的時間輪問題，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

寫在前面

kafka是一個分布式消息中間件，其高可用高吞吐的特點是大數據領域首選的消息中間件，Kafka是分布式消息隊列的順序讀寫文件分段組織串聯起來思想的鼻祖，包括RocketMq這些消息隊列都是借鑒了Kafka早期的架構和設計思路改造而來，所以在架構設計層面，Kafka有非常多值得借鑒的地方。PS：執行流程和代碼來自Kafka0.10.2版本。

引子：

從2個面試題說起，第1個問題，如果一臺機器上有10w個定時任務，如何做到高效觸發？

具體場景是：

有一個APP實時消息通道系統，對每個用戶會維護一個APP到服務器的TCP連接，用來實時收發消息，對這個TCP連接，有這樣一個需求：“如果連續30s沒有請求包（例如登錄，消息，keepalive包），服務端就要將這個用戶的狀態置為離線”。

 其中，單機TCP同時在線量約在10w級別，keepalive請求包較分散大概30s一次，吞吐量約在3000qps。

怎么做？

常用方案使用time定時任務，每秒掃描一次所有連接的集合Map<uid, last_packet_time>，把連接時間（每次有新的請求更新對應連接的連接時間）比當前時間的差值大30s的連接找出來處理。

另一種方案，使用環形隊列法：

（圖1）

三個重要的數據結構：

1）30s超時，就創建一個index從0到30的環形隊列（本質是個數組）

2）環上每一個slot是一個Set<uid>，任務集合

3）同時還有一個Map<uid, index>，記錄uid落在環上的哪個slot里

這樣當有某用戶uid有請求包到達時：

1）從Map結構中，查找出這個uid存儲在哪一個slot里

2）從這個slot的Set結構中，刪除這個uid

3）將uid重新加入到新的slot中，具體是哪一個slot呢 => Current Index指針所指向的上一個slot，因為這個slot，會被timer在30s之后掃描到

4）更新Map，這個uid對應slot的index值

哪些元素會被超時掉呢？

Current Index每秒種移動一個slot，這個slot對應的Set<uid>中所有uid都應該被集體超時！如果最近30s有請求包來到，一定被放到Current Index的前一個slot了，Current Index所在的slot對應Set中所有元素，都是最近30s沒有請求包來到的。

所以，當沒有超時時，Current Index掃到的每一個slot的Set中應該都沒有元素。

兩種方案對比：

方案一每次都要輪詢所有數據，而方案二使用環形隊列只需要輪詢這一刻需要過期的數據，如果沒有數據過期則沒有數據要處理，并且是批量超時，并且由于是環形結構更加節約空間，這很適合高性能場景。

第二個問題：在開發過程中有延遲一定時間的任務要執行，怎么做？

如果不重復造輪子的話，我們的選擇當然是延遲隊列或者Timer。

延遲隊列和在Timer中增 加延時任務采用數組表示的最小堆的數據結構實現，每次放入新元素和移除隊首元素時間復雜度為O(nlog(n))。

 時間輪

方案二所采用的環形隊列，就是時間輪的底層數據結構，它能夠讓需要處理的數據（任務的抽象）集中，在Kafka中存在大量的延遲操作，比如延遲生產、延遲拉取以及延遲刪除等。Kafka并沒有使用JDK自帶的Timer或者DelayQueue來實現延遲的功能，而是基于時間輪自定義了一個用于實現延遲功能的定時器（SystemTimer）。JDK的Timer和DelayQueue插入和刪除操作的平均時間復雜度為O(nlog(n))，并不能滿足Kafka的高性能要求，而基于時間輪可以將插入和刪除操作的時間復雜度都降為O(1)。時間輪的應用并非Kafka獨有，其應用場景還有很多，在Netty、Akka、Quartz、Zookeeper等組件中都存在時間輪的蹤影。

時間輪的數據結構

參考下圖，Kafka中的時間輪（TimingWheel）是一個存儲定時任務的環形隊列，底層采用數組實現，數組中的每個元素可以存放一個定時任務列表（TimerTaskList）。TimerTaskList是一個環形的雙向鏈表，鏈表中的每一項表示的都是定時任務項（TimerTaskEntry），其中封裝了真正的定時任務TimerTask。在Kafka源碼中對這個TimeTaskList是用一個名稱為buckets的數組表示的，所以后面介紹中可能TimerTaskList也會被稱為bucket。

（圖2：圖片來源于《Kafka解惑之時間輪（TimingWheel）》）

針對上圖的幾個名詞簡單解釋下：

tickMs：時間輪由多個時間格組成，每個時間格就是tickMs，它代表當前時間輪的基本時間跨度。

wheelSize：代表每一層時間輪的格數

interval：當前時間輪的總體時間跨度，interval=tickMs × wheelSize

startMs：構造當層時間輪時候的當前時間，第一層的時間輪的startMs是TimeUnit.NANOSECONDS.toMillis(nanoseconds()),上層時間輪的startMs為下層時間輪的currentTime。

currentTime：表示時間輪當前所處的時間，currentTime是tickMs的整數倍（通過currentTime=startMs - (startMs % tickMs來保正currentTime一定是tickMs的整數倍），這個運算類比鐘表中分鐘里65秒分鐘指針指向的還是1分鐘）。currentTime可以將整個時間輪劃分為到期部分和未到期部分，currentTime當前指向的時間格也屬于到期部分，表示剛好到期，需要處理此時間格所對應的TimerTaskList的所有任務。

時間輪中的任務存放

若時間輪的tickMs=1ms，wheelSize=20，那么可以計算得出interval為20ms。初始情況下表盤指針currentTime指向時間格0，此時有一個定時為2ms的任務插入進來會存放到時間格為2的TimerTaskList中。隨著時間的不斷推移，指針currentTime不斷向前推進，過了2ms之后，當到達時間格2時，就需要將時間格2所對應的TimeTaskList中的任務做相應的到期操作。此時若又有一個定時為8ms的任務插入進來，則會存放到時間格10中，currentTime再過8ms后會指向時間格10。如果同時有一個定時為19ms的任務插入進來怎么辦？新來的TimerTaskEntry會復用原來的TimerTaskList，所以它會插入到原本已經到期的時間格1中?？傊?，整個時間輪的總體跨度是不變的，隨著指針currentTime的不斷推進，當前時間輪所能處理的時間段也在不斷后移，總體時間范圍在currentTime和currentTime+interval之間。

時間輪的升降級

如果此時有個定時為350ms的任務該如何處理？直接擴充wheelSize的大小么？Kafka中不乏幾萬甚至幾十萬毫秒的定時任務，這個wheelSize的擴充沒有底線，就算將所有的定時任務的到期時間都設定一個上限，比如100萬毫秒，那么這個wheelSize為100萬毫秒的時間輪不僅占用很大的內存空間，而且效率也會拉低。Kafka為此引入了層級時間輪的概念，當任務的到期時間超過了當前時間輪所表示的時間范圍時，就會嘗試添加到上層時間輪中

（圖3：圖片來源于《Kafka解惑之時間輪（TimingWheel）》）

參考上圖，復用之前的案例，第一層的時間輪tickMs=1ms, wheelSize=20, interval=20ms。第二層的時間輪的tickMs為第一層時間輪的interval，即為20ms。每一層時間輪的wheelSize是固定的，都是20，那么第二層的時間輪的總體時間跨度interval為400ms。以此類推，這個400ms也是第三層的tickMs的大小，第三層的時間輪的總體時間跨度為8000ms。
剛才提到的350ms的任務，不會插入到第一層時間輪，會插入到interval=20*20的第二層時間輪中，具體插入到時間輪的哪個bucket呢？先用350/tickMs(20)=virtualId(17)，然后virtualId(17) %wheelSize (20) = 17，所以350會放在第17個bucket。如果此時有一個450ms后執行的任務，那么會放在第三層時間輪中，按照剛才的計算公式，會放在第0個bucket。第0個bucket里會包含

[400,800)ms的任務。隨著時間流逝，當時間過去了400ms，那么450ms后就要執行的任務還剩下50ms的時間才能執行，此時有一個時間輪降級的操作，將50ms任務重新提交到層級時間輪中，那么此時50ms的任務根據公式會放入第二個時間輪的第2個bucket中，此bucket的時間范圍為[40,60)ms，然后再經過40ms，這個50ms的任務又會被監控到，此時距離任務執行還有10ms，同樣將10ms的任務提交到層級時間輪，此時會加入到第一層時間輪的第10個bucket，所以再經過10ms后，此任務到期，最終執行。

整個時間輪的升級降級操作是不是很類似于我們的時鐘？ 第一層時間輪tickMs=1s, wheelSize=60，interval=1min，此為秒鐘；第二層tickMs=1min，wheelSize=60，interval=1hour，此為分鐘；第三層tickMs=1hour，wheelSize為12，interval為12hours，此為時鐘。而鐘表的指針就對應程序中的currentTime，這個后面分析代碼時候會講到（對這個的理解也是時間輪理解的重點和難點）。

任務添加和驅動時間輪滾動核心流程圖

（圖4）

重點代碼介紹

這是往SystenTimer中添加一個任務

//在Systemtimer中添加一個任務，任務被包裝為一個TimerTaskEntry
private def addTimerTaskEntry(timerTaskEntry: TimerTaskEntry): Unit = {
//先判斷是否可以添加進時間輪中，如果不可以添加進去代表任務已經過期或者任務被取消，注意這里
的timingWheel持有上一層時間輪的引用，所以可能存在遞歸調用
  if (!timingWheel.add(timerTaskEntry)) {    // Already expired or cancelled
    if (!timerTaskEntry.cancelled)     //過期任務直接線程池異步執行掉
      taskExecutor.submit(timerTaskEntry.timerTask)
  }
}

timingWheel添加任務，遞歸添加直到添加該任務進合適的時間輪的bucket中

def add(timerTaskEntry: TimerTaskEntry): Boolean = {
  val expiration = timerTaskEntry.expirationMs  //任務取消
  if (timerTaskEntry.cancelled) {    // Cancelled
    false
  } else if (expiration < currentTime + tickMs) {    // 任務過期后會被執行
    false
  } else if (expiration < currentTime + interval) {//任務過期時間比當前時間輪時間加周期小說明任務
過期時間在本時間輪周期內
    val virtualId = expiration / tickMs    //找到任務對應本時間輪的bucket
    val bucket = buckets((virtualId % wheelSize.toLong).toInt)
    bucket.add(timerTaskEntry)    // Set the bucket expiration time
   //只有本bucket內的任務都過期后才會bucket.setExpiration返回true此時將bucket放入延遲隊列
    if (bucket.setExpiration(virtualId * tickMs)) {     
     //bucket是一個TimerTaskList，它實現了java.util.concurrent.Delayed接口，里面是一個多任務組
     成的鏈表，圖2有說明
      queue.offer(bucket)
    }    true
  } else {    // Out of the interval. Put it into the parent timer
    //任務的過期時間不在本時間輪周期內說明需要升級時間輪，如果不存在則構造上一層時間輪，繼續用
     上一層時間輪添加任務
    if (overflowWheel == null) addOverflowWheel()
    overflowWheel.add(timerTaskEntry)
  }
}

 在本層級時間輪里添加上一層時間輪里的過程，注意的是在下一層時間輪的interval為上一層時間輪的tickMs

private[this] def addOverflowWheel(): Unit = {  
  synchronized {    if (overflowWheel == null) {
      overflowWheel = new TimingWheel(
        tickMs = interval,
        wheelSize = wheelSize,
        startMs = currentTime,
        taskCounter = taskCounter,
        queue
      )
    }
  }
}

驅動時間輪滾動過程：

注意這里會存在一個遞歸，一直驅動時間輪的指針滾動直到時間不足于驅動上層的時間輪滾動。

def advanceClock(timeMs: Long): Unit = {  
if (timeMs >= currentTime + tickMs) {   
  //把當前時間打平為時間輪tickMs的整數倍
    currentTime = timeMs - (timeMs % tickMs)    
    // Try to advance the clock of the overflow wheel if present
    //驅動上層時間輪，這里的傳給上層的currentTime時間是本層時間輪打平過的，但是在上層時間輪還是會繼續打平
    if (overflowWheel != null) overflowWheel.advanceClock(currentTime)
  }
}

 驅動源：

//循環bucket里面的任務列表，一個個重新添加進時間輪，對符合條件的時間輪進行升降級或者執行任務
private[this] val reinsert = (timerTaskEntry: TimerTaskEntry) => addTimerTaskEntry(timerTaskEntry) 
/*
 * Advances the clock if there is an expired bucket. If there isn't any expired bucket when called,
 * waits up to timeoutMs before giving up.
 */
 def advanceClock(timeoutMs: Long): Boolean = {
  var bucket = delayQueue.poll(timeoutMs, TimeUnit.MILLISECONDS) 
   if (bucket != null) {
    writeLock.lock()    try {      while (bucket != null) {        
    //驅動時間輪
        timingWheel.advanceClock(bucket.getExpiration())       
        //循環buckek也就是任務列表，任務列表一個個繼續添加進時間輪以此來升級或者降級時間輪，
        把過期任務找出來執行
        bucket.flush(reinsert)      
         //循環
        //這里就是從延遲隊列取出bucket，bucket是有延遲時間的，取出代表該bucket過期，我們通過
        bucket能取到bucket包含的任務列表
        bucket = delayQueue.poll()
      }
    } finally {
      writeLock.unlock()
    }    true
  } else {    false
  }
}

kafka的延遲隊列使用時間輪實現，能夠支持大量任務的高效觸發，但是在kafka延遲隊列實現方案里還是看到了delayQueue的影子，使用delayQueue是對時間輪里面的bucket放入延遲隊列，以此來推動時間輪滾動，但是基于將插入和刪除操作則放入時間輪中，將這些操作的時間復雜度都降為O(1)，提升效率。Kafka對性能的極致追求讓它把最合適的組件放在最適合的位置。

上述就是小編為大家分享的如何解析Kafka中的時間輪問題了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。