寫在前面

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

引出環形隊列和延遲隊列

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

具體場景是：

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

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

怎么做？

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

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

三個重要的數據結構：

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。

時間輪相關名詞解釋

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中?？傊?#xff0c;整個時間輪的總體跨度是不變的，隨著指針currentTime的不斷推進，當前時間輪所能處理的時間段也在不斷后移，總體時間范圍在currentTime和currentTime+interval之間。

時間輪的升降級

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

參考上圖，復用之前的案例，第一層的時間輪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，這個后面分析代碼時候會講到（對這個的理解也是時間輪理解的重點和難點）。

Kafka中任務添加和驅動時間輪滾動的核心流程：

重點代碼介紹

這是往SystenTimer中添加一個任務

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

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

def add(timerTaskEntry: TimerTaskEntry): Boolean = {val expiration = timerTaskEntry.expirationMs//任務取消if (timerTaskEntry.cancelled) {// Cancelledfalse} else if (expiration < currentTime + tickMs) {// 任務過期后會被執行false} else if (expiration < currentTime + interval) {//任務過期時間比當前時間輪時間加周期小說明任務過期時間在本時間輪周期內val virtualId = expiration / tickMs//找到任務對應本時間輪的bucketval 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對性能的極致追求讓它把最合適的組件放在最適合的位置。

來源：https://www.tuicool.com/articles/fe2quav

總結

以上是生活随笔為你收集整理的跟Kafka学技术系列之时间轮的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。