文章目錄
- 背景
- 時間輪
- 層級時間輪
- 時間輪降級
- kafka中的時間輪
- kafka如何進行時間輪運行
背景
Kafka中存在大量的延時操作,比如延時生產、延時拉取和延時刪除等。Kafka并沒有使用JDK自帶的Timer或DelayQueue來實現延時的功能,而是基于時間輪的概念自定義實現了一個用于延時功能的定時器(SystemTimer)。JDK中Timer和DelayQueue的插入和刪除操作的平均時間復雜度為O(nlogn)并不能滿足Kafka的高性能要求,而基于時間輪可以將插入和刪除操作的時間復雜度都降為O(1)。時間輪的應用并非Kafka獨有,其應用場景還有很多,在Netty Akka、Quartz、ZooKeeper等組件中都存在時間輪的蹤影。
時間輪
如圖所示,Kafka中的時間輪(Timingwheel)是一個存儲定時任務的環形隊列,底層采用數組實現,數組中的每個元素可以存放一個定時任務列表(TimerTaskList)。TimerTaskList是一個環形的雙向鏈表,鏈表中的每一項表示的都是定時任務項(TimerTaskEntry),其中封裝了真正的定時任務(TimerTask)。
時間輪由多個時間格組成,每個時間格代表當前時間輪的基本時間跨度(tickMs)。時間輪的時間格個數是固定的,可用wheelSize來表示,那么整個時間輪的總體時間跨度(interval)可以通過公式tickMs×wheelSize計算得出。時間輪還有一個表盤指針(currentTime),用來表示時間輪當前所處的時間,currentTime是tickMs的整數倍。currentTime可以將整個時間輪劃分為到期部分和未到期部分,currentTime當前指向的時間格也屬于到期部分,表示剛好到期,需要處理此時間格所對應的TimerTaskList中的所有任務。
若時間輪的tickMs為1ms且wheelSize等于20,那么可以計算得出總體時間跨度interval為20ms。初始情況下表盤指針currentTime指向時間格0,此時有一個定時為2ms的任務插進來會存放到時間格為2的TimerTaskList中。隨著時間的不斷推移,指針currentTme不斷向前推進,過了2ms之后,當到達時間格2時,就需要將時間格2對應的TimeTaskList中的任務進行相應的到期操作。此時若又有一個定時為8ms的任務插進來,則會存放到時間格10中,currentTime再過8ms后會指向時間格10。如果同時有一個定時為19ms的任務插進來怎么辦?
新來的TimerTaskEntry會復用原來的TimerTaskList,所以它會插入原本已經到期的時間格1。總之,整個時間輪的總體跨度是不變的,隨著指針currentTime的不斷推進,當前時間輪所能處理的時間段也在不斷后移,總體時間范圍在currenttTime 和 currentTime+interval 之間。
層級時間輪
如果此時有一個定時為350ms的任務該如何處理?直接擴充wheelSize的大小?Kafka中不乏幾萬甚至幾十萬毫秒的定時任務,這個wheelSize的擴充沒有底線,就算將所有的定時任務的到期時間都設定一個上限,比如100萬毫秒,那么這個wheelSize為100萬毫秒的時間輪不僅占用很大的內存空間,而且也會拉低效率。Kafka為此引入了層級時間輪的概念,當任務的到期時間超過了當前時間輪所表示的時間范圍時,就會嘗試添加到上層時間輪中。
如圖所示,復用之前的案例,第一層的時間輪tickMslms、wheelSize=20、interval=20ms。第二層的時間輪的tickMs為第一層時間輪的interval,即20ms。每一層時間輪的wheelSize是固定的,都是20,那么第二層的時間輪的總體時間跨度interval為400ms。以此類推,這個400ms也是第三層的tickMs的大小,第三層的時間輪的總體時間跨度為8000ms。
時間輪降級
對于之前所說的350ms的定時任務,顯然第一層時間輪不能滿足條件,所以就升級到第二層時間輪中,最終被插入第二層時間輪中時間格17所對應的TimerTaskList。如果此時又有一個定時為450ms的任務,那么顯然第二層時間輪也無法滿足條件,所以又升級到第三層時間輪中最終被插入第三層時間輪中時間格1的TimerTaskList。注意到在到期時間為[400ms800ms)區間內的多個任務(比如446ms、455ms和473ms的定時任務)都會被放入第三層時間輪的時間格1,時間格1對應的TimerTaskList的超時時間為400ms。隨著時間的流逝,當此TimerTaskList到期之時,原本定時為450ms的任務還剩下50ms的時間,還不能執行這個任務的到期操作。這里就有一個時間輪降級的操作,會將這個剩余時間為50ms的定時任務重新提交到層級時間輪中,此時第一層時間輪的總體時間跨度不夠,而第二層足夠,所以該任務被放到第二層時間輪到期時間為40ms,60ms)的時間格中再經歷40ms之后,此時這個任務又被“察覺”,不過還剩余10ms,還是不能立即執行到期操作。所以還要再有一次時間輪的降級,此任務被添加到第一層時間輪到期時間為[10ms,11ms)的時間格中,之后再經歷10ms后,此任務真正到期,最終執行相應的到期操作。
設計源于生活。我們常見的鐘表就是一種具有三層結構的時間輪,第一層時間輪
tickMs=lms、wheelSize=60、interval=lmin,此為秒鐘:第二層tickMs=lmin、wheelSize-60、interval=lhour,此為分鐘;第三層tickMs=lhour、wheelSize-12、interval=l2hours,此為時鐘。
kafka中的時間輪
在Kafka中,第一層時間輪的參數同上面的案例一樣:ttickMs=lms、wheelSize-20、interval=20ms,各個層級的wheelSize也固定為20,所以各個層級的tickMs和interval也可以相應地推算出來。Kafka在具體實現時間輪Timingwheel時還有一些小細節:
-
TimingWheel在創建的時候以當前系統時間為第一層時間輪的起始時間JCstartMs)這里的當前系統時間并沒有簡單地調用System.currentTimeMillisO,而是調用了Time.SYSTEM.hiResClockMs,這是因為currentTimeMillisO方法的時間精度依賴于操作系統的具體實現,有些操作系統下并不能達到毫秒級的精度,而Time.SYSTEM.hiResClockMs實質上采用了System.nanoTime(/1000.000來將精度調整到毫秒級。
-
Timingwheel中的每個雙向環形鏈表TimerTaskList都會有一個哨兵節(sentinel),引入哨兵節點可以簡化邊界條件。哨兵節點也稱為啞元節點(dummynode),它是一個附加的鏈表節點,該節點作為第一個節點,它的值域中并不存儲任何東西,只是為了操作的方使而引入的。如果一個鏈表有哨兵節點,那么線性表的第一個元素應該是鏈表的第二個節點。(典型的鏈表數據結構實現)
-
除了第一層時間輪,其余高層時間輪的起始時間(startMs)都設置為創建此層時間輪時前面第一輪的currentTime。每一層的currentTime都必須是tickMs的整數倍,如果不滿足則會將currentTime修剪為tickMs的整數倍,以此與時間輪中的時間格的到期時間范圍對應起來。修剪方法為:currentTimestartMs= startMs -(startMs % tickMs)。 currentTime會隨著時間推移而推進,但不會改變為tickMs的整數倍的既定事實。若某一時刻的時間為timeMs,那么此時時間輪的currentTime
=timeMs-(timeMs%tickMs),時間每推進一次,每個層級的時間輪的currentTime都會依據此公式執行推進。 -
Kafka中的定時器只需持有Timingwheel的第一層時間輪的引用,并不會直接持有其他高層的時間輪,但每一層時間輪都會有一個引用(overfowwheel)指向更高一層的應用,以此層級調用可以實現定時器間接持有各個層級時間輪的引用。
kafka如何進行時間輪運行
上面我們說過“隨著時間的流逝”或“隨著時間的推移”,那么在Kafka中到底是怎么推進時間的呢?類似采用JDK中的scheduleAtFixedRate來每秒推進時間輪?顯然這樣并不合理,Timingwheel1也失去了大部分意義。
Kafka中的定時器借了JDK中的DelayQueue來協助推進時間輪。具體做法是對于每個使用到的TimerTaskList者都加入DelayQueue,每個用到的TimerTaskList”特指非哨兵節點的定時任務項TimerTaskEntry對應的TimerTaskListoDelayQueue會根據TimerTaskList對應的超時時間expiration來排序,最短expiration的TimerTaskList會被排在DelayOueue的隊頭。Kafka中會有一個線程來獲取DelayQueue中到期的任務列表,有意思的是這個線程所對應的名稱叫作“ExpiredOperationReaper”,可以直譯為“過期操作收割機”。當“收割機”線程獲取DelayQueue中超時的任務列表TimerTaskList之后,既可以根據TimerTaskList的expiration來推進時間輪的時間,也可以就獲取的TimerTaskList執行相應的操作,對里面的TimerTaskEntry該執行過期操作的就執行過期操作,該降級時間輪的就降級時間輪。
看到這里或許會感到困惑,開頭明確指明的DelayQueue不適合Kafka這種高性能要求的定時任務,為何這里還要引入DelayQueue呢?注意對定時任務項TimerTaskEntry的插入和刪除操作而言,Timingwheel時間復雜度為0(I),性能高出DelayQueue很多,如果直接將TimerTaskEntry插入DelayQueue,那么性能顯然難以支撐。就算我們根據一定的規則將若干TimerTaskEntry劃分到TimerTaskList這個組中,然后將TimerTaskList插入DelayQueue,如果在TimerTaskList中又要多添加一個TimerTaskEntry時該如何處理呢?對DelayQueue而言,這類操作顯然變得力不從心。
到這里可以發現,Kafka中的Timingwheel專門用來執行插入和刪除TimerTaskEntry的操作,而DelayQueue專門負責時間推進的任務。試想一下,DelayQueuee中的第一個超時任務列表的expiration為200ms,第二個超時任務為840ms,這里獲取DelayQueue的隊頭只需要O(1)的時間復雜度(獲取之后DelayQueue內部才會再次切換出新的隊頭)。如果采用每秒定時推進,那么獲取第一個超時的任務列表時執行的200次推進中有199次屬于“空推進”,而獲取第二個超時任務時又需要執行639次“空推進”,這樣會無故空耗機器的性能資源,這里采用DelayQueue來輔助以少量空間換時間,從而做到了“精準推進”。Kafka中的定時器真可謂“知人善用”,用Timingwheel做最擅長的任務添加和刪除操作,而用DelayQueue做最擅長的時間推進工作,兩者相輔相成。