延遲隊列的需求各位應(yīng)該在日常開發(fā)的場景中經(jīng)常碰到。比如：

• 用戶登錄之后5分鐘給用戶做分類推送；
• 用戶多少天未登錄給用戶做召回推送；
• 定期檢查用戶當前退款賬單是否被商家處理等等場景。

一般這種場景和定時任務(wù)還是有很大的區(qū)別，定時任務(wù)是你知道任務(wù)多久該跑一次或者什么時候只跑一次，這個時間是確定的。延遲隊列是當某個事件發(fā)生的時候需要延遲多久觸發(fā)配套事件，引子事件發(fā)生的時間不是固定的。

業(yè)界目前也有很多實現(xiàn)方案，單機版的方案就不說了，現(xiàn)在也沒有哪個公司還是單機版的服務(wù)，今天我們一一探討各種方案的大致實現(xiàn)。

1. Redis zset

這個方案比較常用，簡單有效。利用 Redis 的 sorted set 結(jié)構(gòu)，使用 timeStamp 作為 score，比如你的任務(wù)是要延遲5分鐘，那么就在當前時間上加5分鐘作為 score ，輪詢?nèi)蝿?wù)每秒只輪詢 score 大于當前時間的 key即可，如果任務(wù)支持有誤差，那么當沒有掃描到有效數(shù)據(jù)的時候可以休眠對應(yīng)時間再繼續(xù)輪詢。

方案優(yōu)劣：

優(yōu)點：

簡單實用，一針見血。

缺點：

1. 單個 zset 肯定支持不了太大的數(shù)據(jù)量，如果你有幾百萬的延遲任務(wù)需求，大哥我還是勸你換一個方案；
2. 定時器輪詢方案可能會有異常終止的情況需要自己處理，同時消息處理失敗的回滾方案，您也要自己處理。

所以，sorted set 的方案并不是一個成熟的方案，他只是一個快速可供落地的方案。

2. RabbitMQ隊列

下面說一個可以落地的方案，這個方案也被大多數(shù)目前在架構(gòu)中使用了 RabbitMQ 的項目組使用。不好的一點就是，捆綁 RabbitMQ，當你的架構(gòu)方案是要用別的 MQ 替換 RabbitMQ 的時候，你就蛋疼了（我現(xiàn)在正在經(jīng)歷）。

RabbitMQ 有兩個特性，一個是 Time-To-Live Extensions，另一個是 Dead Letter Exchanges。

Time-To-Live Extensions

RabbitMQ允許我們?yōu)橄⒒蛘哧犃性O(shè)置TTL（time to live），也就是過期時間。TTL表明了一條消息可在隊列中存活的最大時間，單位為毫秒。也就是說，當某條消息被設(shè)置了TTL或者當某條消息進入了設(shè)置了TTL的隊列時，這條消息會在經(jīng)過TTL秒后 “死亡”，成為Dead Letter。如果既配置了消息的TTL，又配置了隊列的TTL，那么較小的那個值會被取用。

Dead Letter Exchanges

在 RabbitMQ 中，一共有三種消息的 “死亡” 形式：

• 消息被拒絕。通過調(diào)用 basic.reject 或者 basic.nack 并且設(shè)置的 requeue 參數(shù)為 false；
• 消息因為設(shè)置了TTL而過期；
• 隊列達到最大長度。
  

DLX同一般的 Exchange 沒有區(qū)別，它能在任何的隊列上被指定，實際上就是設(shè)置某個隊列的屬性。當隊列中有 DLX 消息時，RabbitMQ就會自動的將 DLX 消息重新發(fā)布到設(shè)置的 Exchange 中去，進而被路由到另一個隊列，publish 可以監(jiān)聽這個隊列中消息做相應(yīng)的處理。

由上簡介大家可以看出，RabbitMQ本身是不支持延遲隊列的，只是他的特性讓勤勞的 中國脫發(fā)群體 急中生智（為了完成任務(wù)）弄出了這么一套可用的方案。

可用的方案就是：

1. 如果有事件需要延遲那么將該事件發(fā)送到MQ 隊列中，為需要延遲的消息設(shè)置一個TTL；
2. TTL到期后就會自動進入設(shè)置好的DLX，然后由DLX轉(zhuǎn)發(fā)到配置好的實際消費隊列；
3. 消費該隊列的延遲消息，處理事件。
   

方案優(yōu)劣：

優(yōu)點：

大品牌組件，用的放心。如果面臨大數(shù)據(jù)量需求可以很容易的橫向擴展，同時消息支持持久化，有問題可回滾。

缺點：

1. 配置麻煩，額外增加一個死信交換機和一個死信隊列的配置；
2. RabbitMQ 是一個消息中間件，TTL 和 DLX 只是他的一個特性，將延遲隊列綁定在一個功能軟件的某一個特性上，可能會有風險。不要杠，當你們組不用 RabbitMQ 的時候遷移很痛苦；
3. 消息隊列具有先進先出的特點，如果第一個進入隊列的消息 A 的延遲是10分鐘，第二個進入隊列的消息B 的延遲是5分鐘，期望的是誰先到 TTL誰先出，但是事實是B已經(jīng)到期了，而還要等到 A 的延遲10分鐘結(jié)束A先出之后，B 才能出。所以在設(shè)計的時候需要考慮不同延遲的消息要放到不同的隊列。另外該問題官方已經(jīng)給出了插件來支持：插件地址。
   

3. 基于 Netty#HashedWheelTimer類方法的實現(xiàn)

HashedWheelTimer 是 Netty 中 的一個基礎(chǔ)工具類，主要用來高效處理大量定時任務(wù)，且任務(wù)對時間精度要求相對不高， 在Netty 中的應(yīng)用場景就是連接超時或者任務(wù)處理超時，一般都是操作比較快速的任務(wù)，缺點是內(nèi)存占用相對較高。

算法思想

HashedWheelTimer 主要還是一個 DelayQueue 和一個時間輪算法組合。

Hash Wheel Timer是一個環(huán)形結(jié)構(gòu)，可以想象成時鐘，分為很多格子，一個格子代表一段時間（越短Timer精度越高），并用一個List保存在該格子上到期的所有任務(wù)。同時一個指針隨著時間流逝一格一格轉(zhuǎn)動，并執(zhí)行對應(yīng)List中所有到期的任務(wù)。

以上圖為例，假設(shè)一個格子是1s，則整個時間輪能表示的時間段16s。當前任務(wù)指向格子2，表明在第2s的時候有任務(wù)需要執(zhí)行。任務(wù)列表中有兩個任務(wù)，每個任務(wù)前面的數(shù)字表示圈數(shù)。2表示當走到第2圈的時候才會執(zhí)行，那么整個任務(wù)的真正執(zhí)行時間其實是在12s之后執(zhí)行，即第二圈走到2的時候。每推進一格，對應(yīng)的每一個 slot 中的round數(shù)都要減一。整體算法就是這么個邏輯。

時間輪設(shè)計要點：

• tick，一次時間推進，每次推進會檢查/執(zhí)行超時任務(wù)；
• tickDuration，時間輪推進的最小單元，每隔 tickDuration 會有一次 tick，它決定了時間輪的精確程度；
• bucket（ticksPerWheel），上圖中的每一隔就是一個bucket，表示一個時間輪可以有多少個tick，它是存儲任務(wù)的最小單元；
• 上層時間輪的 tickDuration 是下層時間輪的表示時間的最大范圍，即：父 tickDuration = 子 tickDuration * 子 bucket 。

需要注意的是，這種方式任務(wù)是串行執(zhí)行的。意味著你如果在時間輪中執(zhí)行任務(wù)且任務(wù)耗時較長，將會出現(xiàn)調(diào)度超時或者任務(wù)堆積的情況。所以要將任務(wù)的執(zhí)行異步化。

算法的要點：

1. 任務(wù)并不是直接放在格子中的，而是維護了一個雙向鏈表，這種數(shù)據(jù)結(jié)構(gòu)非常便于插入和移除；
2. 新添加的任務(wù)并不直接放入格子，而是先放入一個隊列中，這是為了避免多線程插入任務(wù)的沖突。在每個tick運行任務(wù)之前由worker線程自動對任務(wù)進行歸集和分類，插入到對應(yīng)的槽位里面。
   

Netty 使用數(shù)組 + 雙向鏈表的方式來組織時間輪，對于添加/取消操作僅做了記錄，真正的操作實際發(fā)生在下一個tick。時間的推進是獨立的線程在做，該線程同時也負責過期任務(wù)的執(zhí)行等操作，可簡單認為此步驟操作為O(n)，因為推進線程需要完全遍歷timeouts、cancelledTimeouts與bucket鏈表，在遍歷timeouts時，Netty為了避免任務(wù)過多，所以限制每次最多遍歷10萬個，也就是說，一個tick只能規(guī)劃10萬個任務(wù)，當任務(wù)量過大時，會存在超時任務(wù)執(zhí)行時間延遲的現(xiàn)象。

方案優(yōu)劣：

優(yōu)點：

實現(xiàn)比較優(yōu)雅。效率高。

缺點：

無法實現(xiàn)HA和橫向擴展，要么就使用多個時間輪。
最重要的是，實現(xiàn)也比較復(fù)雜，開發(fā)者需要考慮所有可能的情況。
目前我了解到的延遲隊列在生產(chǎn)環(huán)境下有如上三種實現(xiàn)方式，每一種都有人在使用。當然沒有最好的只有最適合的，你覺得 redis 能滿足需求，就按照最簡單的來，你要是有充足的開發(fā)周期，你也可以實現(xiàn)時間輪展現(xiàn)實力。

需求千萬種，變化就一種：給時間都能做。

以上就是JAVA 實現(xiàn)延遲隊列的方法的詳細內(nèi)容，更多關(guān)于JAVA 實現(xiàn)延遲隊列的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論