最近看 Kafka 看到了時(shí)間輪算法,記得以前看 Netty 也看到過這玩意,沒太過關(guān)注。今天就來看看時(shí)間輪到底是什么東西。
為什么要用時(shí)間輪算法來實(shí)現(xiàn)延遲操作?
延時(shí)操作 JAVA 不是提供了 Timer 么?
還有 DelayQueue 配合線程池或者 ScheduledThreadPool 不香嗎?
我們先來簡(jiǎn)單看看 Timer、DelayQueue 和 ScheduledThreadPool 的相關(guān)實(shí)現(xiàn),看看它們是如何實(shí)現(xiàn)延時(shí)任務(wù)的,源碼之下無秘密。再來剖析下為何 Netty 和 Kafka 特意實(shí)現(xiàn)了時(shí)間輪來處理延遲任務(wù)。
如果在手機(jī)上閱讀其實(shí)純看字也行,不用看代碼,我都會(huì)先用文字描述清楚。不過電腦上看效果更佳。
Timer
Timer 可以實(shí)現(xiàn)延時(shí)任務(wù),也可以實(shí)現(xiàn)周期性任務(wù)。我們先來看看 Timer 核心屬性和構(gòu)造器。
核心就是一個(gè)優(yōu)先隊(duì)列和封裝的執(zhí)行任務(wù)的線程,從這我們也可以看到一個(gè) Timer 只有一個(gè)線程執(zhí)行任務(wù)。
再來看看如何實(shí)現(xiàn)延時(shí)和周期性任務(wù)的。我先簡(jiǎn)單的概括一下,首先維持一個(gè)小頂堆,即最快需要執(zhí)行的任務(wù)排在優(yōu)先隊(duì)列的第一個(gè),根據(jù)堆的特性我們知道插入和刪除的時(shí)間復(fù)雜度都是 O(logn)。
然后 TimerThread 不斷地拿排著的第一個(gè)任務(wù)的執(zhí)行時(shí)間和當(dāng)前時(shí)間做對(duì)比。如果時(shí)間到了先看看這個(gè)任務(wù)是不是周期性執(zhí)行的任務(wù),如果是則修改當(dāng)前任務(wù)時(shí)間為下次執(zhí)行的時(shí)間,如果不是周期性任務(wù)則將任務(wù)從優(yōu)先隊(duì)列中移除。最后執(zhí)行任務(wù)。如果時(shí)間還未到則調(diào)用 wait() 等待。
再看下圖,整理下流程。
流程知道了再對(duì)著看下代碼,這塊就差不多了。看代碼不爽的可以跳過代碼部分,影響不大。
先來看下 TaskQueue,就簡(jiǎn)單看下插入任務(wù)的過程,就是個(gè)普通的堆插入操作。
再來看看 TimerThread 的 run 操作。
小結(jié)一下
可以看出 Timer 實(shí)際就是根據(jù)任務(wù)的執(zhí)行時(shí)間維護(hù)了一個(gè)優(yōu)先隊(duì)列,并且起了一個(gè)線程不斷地拉取任務(wù)執(zhí)行。
有什么弊端呢?
首先優(yōu)先隊(duì)列的插入和刪除的時(shí)間復(fù)雜度是O(logn),當(dāng)數(shù)據(jù)量大的時(shí)候,頻繁的入堆出堆性能有待考慮。
并且是單線程執(zhí)行,那么如果一個(gè)任務(wù)執(zhí)行的時(shí)間過久則會(huì)影響下一個(gè)任務(wù)的執(zhí)行時(shí)間(當(dāng)然你任務(wù)的run要是異步執(zhí)行也行)。
并且從代碼可以看到對(duì)異常沒有做什么處理,那么一個(gè)任務(wù)出錯(cuò)的時(shí)候會(huì)導(dǎo)致之后的任務(wù)都無法執(zhí)行。
ScheduledThreadPoolExecutor
在說 ScheduledThreadPoolExecutor 之前我們?cè)倏聪?Timer 的注釋,注釋可都是干貨千萬不要錯(cuò)過。我做了點(diǎn)修改,突出了下重點(diǎn)。
Java 5.0 introduced ScheduledThreadPoolExecutor, It is effectively a more versatile replacement for the Timer, it allows multiple service threads. Configuring with one thread makes it equivalent to Timer。
簡(jiǎn)單翻譯下:1.5 引入了 ScheduledThreadPoolExecutor,它是一個(gè)具有更多功能的 Timer 的替代品,允許多個(gè)服務(wù)線程。如果設(shè)置一個(gè)服務(wù)線程和 Timer 沒啥差別。
從注釋看出相對(duì)于 Timer ,可能就是單線程跑任務(wù)和多線程跑任務(wù)的區(qū)別。我們來看下。
繼承了 ThreadPoolExecutor,實(shí)現(xiàn)了 ScheduledExecutorService。可以定性操作就是正常線程池差不多了。區(qū)別就在于兩點(diǎn),一個(gè)是 ScheduledFutureTask ,一個(gè)是 DelayedWorkQueue。
其實(shí) DelayedWorkQueue 就是優(yōu)先隊(duì)列,也是利用數(shù)組實(shí)現(xiàn)的小頂堆。而 ScheduledFutureTask 繼承自 FutureTask 重寫了 run 方法,實(shí)現(xiàn)了周期性任務(wù)的需求。
小結(jié)一下
ScheduledThreadPoolExecutor 大致的流程和 Timer 差不多,也是維護(hù)一個(gè)優(yōu)先隊(duì)列,然后通過重寫 task 的 run 方法來實(shí)現(xiàn)周期性任務(wù),主要差別在于能多線程運(yùn)行任務(wù),不會(huì)單線程阻塞。
并且 Java 線程池的設(shè)定是 task 出錯(cuò)會(huì)把錯(cuò)誤吃了,無聲無息的。因此一個(gè)任務(wù)出錯(cuò)也不會(huì)影響之后的任務(wù)。
DelayQueue
Java 中還有個(gè)延遲隊(duì)列 DelayQueue,加入延遲隊(duì)列的元素都必須實(shí)現(xiàn) Delayed 接口。延遲隊(duì)列內(nèi)部是利用 PriorityQueue 實(shí)現(xiàn)的,所以還是利用優(yōu)先隊(duì)列!Delayed 接口繼承了Comparable 因此優(yōu)先隊(duì)列是通過 delay 來排序的。
然后我們?cè)賮砜聪卵舆t隊(duì)列是如何獲取元素的。
小結(jié)一下
也是利用優(yōu)先隊(duì)列實(shí)現(xiàn)的,元素通過實(shí)現(xiàn) Delayed 接口來返回延遲的時(shí)間。不過延遲隊(duì)列就是個(gè)容器,需要其他線程來獲取和執(zhí)行任務(wù)。
這下是搞明白了 Timer 、ScheduledThreadPool 和 DelayQueue,總結(jié)的說下它們都是通過優(yōu)先隊(duì)列來獲取最早需要執(zhí)行的任務(wù),因此插入和刪除任務(wù)的時(shí)間復(fù)雜度都為O(logn),并且 Timer 、ScheduledThreadPool 的周期性任務(wù)是通過重置任務(wù)的下一次執(zhí)行時(shí)間來完成的。
問題就出在時(shí)間復(fù)雜度上,插入刪除時(shí)間復(fù)雜度是O(logn),那么假設(shè)頻繁插入刪除次數(shù)為 m,總的時(shí)間復(fù)雜度就是O(mlogn),這種時(shí)間復(fù)雜度滿足不了 Kafka 這類中間件對(duì)性能的要求,而時(shí)間輪算法的插入刪除時(shí)間復(fù)雜度是O(1)。我們來看看時(shí)間輪算法是如何實(shí)現(xiàn)的。
時(shí)間輪算法
俗話說藝術(shù)源于生活,技術(shù)也能從日常生活中找到靈感。咱們先來看塊表,嗯金色的表。
都看清楚了吧,時(shí)間輪就是和手表時(shí)鐘很相似的存在。時(shí)間輪用環(huán)形數(shù)組實(shí)現(xiàn),數(shù)組的每個(gè)元素可以稱為槽,和 HashMap一樣稱呼。
槽的內(nèi)部用雙向鏈表存著待執(zhí)行的任務(wù),添加和刪除的鏈表操作時(shí)間復(fù)雜度都是 O(1),槽位本身也指代時(shí)間精度,比如一秒掃一個(gè)槽,那么這個(gè)時(shí)間輪的最高精度就是 1 秒。
也就是說延遲 1.2 秒的任務(wù)和 1.5 秒的任務(wù)會(huì)被加入到同一個(gè)槽中,然后在 1 秒的時(shí)候遍歷這個(gè)槽中的鏈表執(zhí)行任務(wù)。
從圖中可以看到此時(shí)指針指向的是第一個(gè)槽,一共有八個(gè)槽0~7,假設(shè)槽的時(shí)間單位為 1 秒,現(xiàn)在要加入一個(gè)延時(shí) 5 秒的任務(wù),計(jì)算方式就是 5 % 8 + 1 = 6,即放在槽位為 6,下標(biāo)為 5 的那個(gè)槽中。更具體的就是拼到槽的雙向鏈表的尾部。
然后每秒指針順時(shí)針移動(dòng)一格,這樣就掃到了下一格,遍歷這格中的雙向鏈表執(zhí)行任務(wù)。然后再循環(huán)繼續(xù)。
可以看到插入任務(wù)從計(jì)算槽位到插入鏈表,時(shí)間復(fù)雜度都是O(1)。那假設(shè)現(xiàn)在要加入一個(gè)50秒后執(zhí)行的任務(wù)怎么辦?這槽好像不夠啊?難道要加槽嘛?和HashMap一樣擴(kuò)容?
不是的,常見有兩種方式,一種是通過增加輪次的概念。50 % 8 + 1 = 3,即應(yīng)該放在槽位是 3,下標(biāo)是 2 的位置。然后 (50 - 1) / 8 = 6,即輪數(shù)記為 6。也就是說當(dāng)循環(huán) 6 輪之后掃到下標(biāo)的 2 的這個(gè)槽位會(huì)觸發(fā)這個(gè)任務(wù)。Netty 中的 HashedWheelTimer 使用的就是這種方式。
還有一種是通過多層次的時(shí)間輪,這個(gè)和我們的手表就更像了,像我們秒針走一圈,分針走一格,分針走一圈,時(shí)針走一格。
多層次時(shí)間輪就是這樣實(shí)現(xiàn)的。假設(shè)上圖就是第一層,那么第一層走了一圈,第二層就走一格。
可以得知第二層的一格就是8秒,假設(shè)第二層也是 8 個(gè)槽,那么第二層走一圈,第三層走一格,可以得知第三層一格就是 64 秒。
那么一格三層,每層8個(gè)槽,一共 24 個(gè)槽時(shí)間輪就可以處理最多延遲 512 秒的任務(wù)。
而多層次時(shí)間輪還會(huì)有降級(jí)的操作,假設(shè)一個(gè)任務(wù)延遲 500 秒執(zhí)行,那么剛開始加進(jìn)來肯定是放在第三層的,當(dāng)時(shí)間過了 436 秒后,此時(shí)還需要 64 秒就會(huì)觸發(fā)任務(wù)的執(zhí)行,而此時(shí)相對(duì)而言它就是個(gè)延遲 64 秒后的任務(wù),因此它會(huì)被降低放在第二層中,第一層還放不下它。
再過個(gè) 56 秒,相對(duì)而言它就是個(gè)延遲 8 秒后執(zhí)行的任務(wù),因此它會(huì)再被降級(jí)放在第一層中,等待執(zhí)行。
降級(jí)是為了保證時(shí)間精度一致性。Kafka內(nèi)部用的就是多層次的時(shí)間輪算法。
Netty中的時(shí)間輪
在 Netty 中時(shí)間輪的實(shí)現(xiàn)類是 HashedWheelTimer,代碼中的 wheel 就是上圖畫的循環(huán)數(shù)組,mask 的設(shè)計(jì)和HashMap一樣,通過限制數(shù)組的大小為2的次方,利用位運(yùn)算來替代取模運(yùn)算,提高性能。tickDuration 就是每格的時(shí)間即精度。可以看到配備了一個(gè)工作線程來處理任務(wù)的執(zhí)行。
接下來我們?cè)賮砜纯慈蝿?wù)是如何添加的。
可以看到任務(wù)并沒有直接添加到時(shí)間輪中,而是先入了一個(gè) mpsc 隊(duì)列,我簡(jiǎn)單說下 mpsc 是 JCTools 中的并發(fā)隊(duì)列,用在多個(gè)生產(chǎn)者可同時(shí)訪問隊(duì)列,但只有一個(gè)消費(fèi)者會(huì)訪問隊(duì)列的情況。篇幅有限,有興趣的朋友自行了解實(shí)現(xiàn)。
然后我們?cè)賮砜纯垂ぷ骶€程是如何運(yùn)作的。
很直觀沒什么花頭,我們先來看看 waitForNextTick,是如何得到下一次執(zhí)行時(shí)間的。
簡(jiǎn)單的說就是通過 tickDuration 和此時(shí)已經(jīng)滴答的次數(shù)算出下一次需要檢查的時(shí)間,時(shí)候未到就sleep等著。
再來看下任務(wù)如何入槽的。
注釋的很清楚了,實(shí)現(xiàn)也和上述分析的一致。
最后再來看下如何執(zhí)行的。
就是通過輪數(shù)和時(shí)間雙重判斷,執(zhí)行完了移除任務(wù)。
小結(jié)一下
總體上看 Netty 的實(shí)現(xiàn)就是上文說的時(shí)間輪通過輪數(shù)的實(shí)現(xiàn),完全一致。可以看出時(shí)間精度由 TickDuration 把控,并且工作線程的除了處理執(zhí)行到時(shí)的任務(wù)還做了其他操作,因此任務(wù)不一定會(huì)被精準(zhǔn)的執(zhí)行。
而且任務(wù)的執(zhí)行如果不是新起一個(gè)線程,或者將任務(wù)扔到線程池執(zhí)行,那么耗時(shí)的任務(wù)會(huì)阻塞下個(gè)任務(wù)的執(zhí)行。
并且會(huì)有很多無用的 tick 推進(jìn),例如 TickDuration 為1秒,此時(shí)就一個(gè)延遲350秒的任務(wù),那就是有349次無用的操作。
但是從另一面來看,如果任務(wù)都執(zhí)行很快(當(dāng)然你也可以異步執(zhí)行),并且任務(wù)數(shù)很大,通過分批執(zhí)行,并且增刪任務(wù)的時(shí)間復(fù)雜度都是O(1)來說。時(shí)間輪還是比通過優(yōu)先隊(duì)列實(shí)現(xiàn)的延時(shí)任務(wù)來的合適些。
Kafka 中的時(shí)間輪
上面我們說到 Kafka 中的時(shí)間輪是多層次時(shí)間輪實(shí)現(xiàn),總的而言實(shí)現(xiàn)和上述說的思路一致。不過細(xì)節(jié)有些不同,并且做了點(diǎn)優(yōu)化。
先看看添加任務(wù)的方法。在添加的時(shí)候就設(shè)置任務(wù)執(zhí)行的絕對(duì)時(shí)間。
那么時(shí)間輪是如何推動(dòng)的呢?Netty 中是通過固定的時(shí)間間隔掃描,時(shí)候未到就等待來進(jìn)行時(shí)間輪的推動(dòng)。上面我們分析到這樣會(huì)有空推進(jìn)的情況。
而 Kafka 就利用了空間換時(shí)間的思想,通過 DelayQueue,來保存每個(gè)槽,通過每個(gè)槽的過期時(shí)間排序。這樣擁有最早需要執(zhí)行任務(wù)的槽會(huì)有優(yōu)先獲取。如果時(shí)候未到,那么 delayQueue.poll 就會(huì)阻塞著,這樣就不會(huì)有空推進(jìn)的情況發(fā)送。
我們來看下推進(jìn)的方法。
從上面的 add 方法我們知道每次對(duì)比都是根據(jù)expiration < currentTime + interval 來進(jìn)行對(duì)比的,而advanceClock 就是用來推進(jìn)更新 currentTime 的。
小結(jié)一下
Kafka 用了多層次時(shí)間輪來實(shí)現(xiàn),并且是按需創(chuàng)建時(shí)間輪,采用任務(wù)的絕對(duì)時(shí)間來判斷延期,并且對(duì)于每個(gè)槽(槽內(nèi)存放的也是任務(wù)的雙向鏈表)都會(huì)維護(hù)一個(gè)過期時(shí)間,利用 DelayQueue 來對(duì)每個(gè)槽的過期時(shí)間排序,來進(jìn)行時(shí)間的推進(jìn),防止空推進(jìn)的存在。
每次推進(jìn)都會(huì)更新 currentTime 為當(dāng)前時(shí)間戳,當(dāng)然做了點(diǎn)微調(diào)使得 currentTime 是 tickMs 的整數(shù)倍。并且每次推進(jìn)都會(huì)把能降級(jí)的任務(wù)重新插入降級(jí)。
可以看到這里的 DelayQueue 的元素是每個(gè)槽,而不是任務(wù),因此數(shù)量就少很多了,這應(yīng)該是權(quán)衡了對(duì)于槽操作的延時(shí)隊(duì)列的時(shí)間復(fù)雜度與空推進(jìn)的影響。
總結(jié)
首先介紹了 Timer、DelayQueue 和 ScheduledThreadPool,它們都是基于優(yōu)先隊(duì)列實(shí)現(xiàn)的,O(logn) 的時(shí)間復(fù)雜度在任務(wù)數(shù)多的情況下頻繁的入隊(duì)出隊(duì)對(duì)性能來說有損耗。因此適合于任務(wù)數(shù)不多的情況。
Timer 是單線程的會(huì)有阻塞的風(fēng)險(xiǎn),并且對(duì)異常沒有做處理,一個(gè)任務(wù)出錯(cuò) Timer 就掛了。而 ScheduledThreadPool 相比于 Timer 首先可以多線程來執(zhí)行任務(wù),并且線程池對(duì)異常做了處理,使得任務(wù)之間不會(huì)有影響。
并且 Timer 和 ScheduledThreadPool 可以周期性執(zhí)行任務(wù)。而 DelayQueue 就是個(gè)具有優(yōu)先級(jí)的阻塞隊(duì)列。
對(duì)比而言時(shí)間輪更適合任務(wù)數(shù)很大的延時(shí)場(chǎng)景,它的任務(wù)插入和刪除時(shí)間復(fù)雜度都為O(1)。對(duì)于延遲超過時(shí)間輪所能表示的范圍有兩種處理方式,一是通過增加一個(gè)字段-輪數(shù),Netty 就是這樣實(shí)現(xiàn)的。二是多層次時(shí)間輪,Kakfa 是這樣實(shí)現(xiàn)的。
相比而言 Netty 的實(shí)現(xiàn)會(huì)有空推進(jìn)的問題,而 Kafka 采用 DelayQueue 以槽為單位,利用空間換時(shí)間的思想解決了空推進(jìn)的問題。
可以看出延遲任務(wù)的實(shí)現(xiàn)都不是很精確的,并且或多或少都會(huì)有阻塞的情況,即使你異步執(zhí)行,線程不夠的情況下還是會(huì)阻塞。
巨人的肩膀
《深入理解Kafka:核心設(shè)計(jì)與實(shí)踐原理》
https://www.cnblogs.com/luozhiyun/p/12075326.html