點擊上方藍色字體，選擇“設為星標”

延遲隊列，它是一種帶有延遲功能的消息隊列，目前工作中有幾處需延時處理的應用場景。

kafka

考慮前提：由于項目代碼與業(yè)務方交互大多采用 kafka，所以想是否能自己集成一個 kafka 延遲隊列，直接提供延遲功能，更方便使用。

大致思路：借鑒 rocketMQ 延遲隊列設計思想，創(chuàng)建多個topic 用于處理不同的延遲消息，例如延遲一分鐘的任務消息，讓 topic為 delay-minutes-1 進行處理。

• 發(fā)送延遲消息時不直接發(fā)到目標topic，而是發(fā)到一個用于處理延遲消息的topic，例如 delay-minutes-1

• 寫一段代碼定時拉取 delay-minutes-1 中的消息，將滿足的消息發(fā)到真正的目標主題里。

流程圖:

解決問題：如何讓延遲消息等待一段時間才發(fā)送到真正的topic里面？

答：KafkaConsumer 提供了暫停和恢復的 API 函數(shù)，當消費者發(fā)現(xiàn)不滿足消費時間條件時，可以先暫停消費者，并把消費偏移量移動到上次位置，進行等待下次消費。

缺點：kafka內(nèi)部改造復雜度較高，由于要使 consumer 進行 pause，還需要額外的做一些健康檢查操作，在狀態(tài)不對時可以報警或者重啟。另外，不支持靈活設置延時時間。

rocketMQ

考慮前提：底層代碼已經(jīng)全部封裝好，直接使用，不用關心底層代碼，可以實現(xiàn)與業(yè)務進行解藕。

大致原理思路：

• RocketMQ將延時隊列的延時時間分為 18 個級別，在發(fā)送 MQ消息的時候只需要設置 delayLevel，把每種延遲時間段的消息放到同一個隊列中

• 通過一個定時器進行輪詢這些隊列，查看消息是否到期

流程圖：

缺點：

• 使用中間件，盡可能的需要熟讀底層源碼，以便后續(xù)出現(xiàn)問題，快速跟蹤定位。還有能找到適合的擴展點。

• 定時器采用的timer是單線程運行，如果延遲消息數(shù)量很大的話，可能造成消息到期也沒有發(fā)送出去的情況。

redis

考慮前提：Redisson延時隊列，代碼redis已經(jīng)封裝好，可以直接拿來用。redisson.getBlockingQueue() 和 Redission.getDelayQueue()

大致原理思路：https://zhuanlan.zhihu.com/p/343811173

三個核心集合結構：

延時隊列：數(shù)據(jù)入隊的隊列

目標 blocking 隊列 ：到期數(shù)據(jù)待consume

timeoutSet 過期時間zset：分數(shù)值為timeout，輔佐判斷元素是否過期。?

實現(xiàn) Timer ：

運用了 redis 的 sub/pub 功能，當有數(shù)據(jù)put的時候，先把它放到一個zset集合，同時發(fā)布訂閱的key，發(fā)布內(nèi)容為數(shù)據(jù)到期的timeout，此時客戶端開啟了一個延時任務（HashedWheelTimer），到了時間，從zset分頁取出到期了的數(shù)據(jù)，放入 blocking 隊列中。

缺點：

• 采用 sub/pub 機制的時候，可能會造成多個客戶端同時開啟一個時間段的延時任務，重復執(zhí)行，也會有并發(fā)的安全問題，因為涉及的要數(shù)據(jù)加入阻塞隊列，和將當前數(shù)據(jù)從zset移除操作。

• 默認是數(shù)據(jù)量小的時候比較穩(wěn)定，數(shù)據(jù)量一大就需要構建 cluster模式，這一塊需要自己開發(fā)

有贊的延時隊列

https://tech.youzan.com/queuing_delay/

實現(xiàn)邏輯圖

各個組件含義：

job ：需要異步處理的任務，是最基本單元，其中屬性包含，自定義唯一jobid，topic任務類型，delayTime任務執(zhí)行時間，ttrtime執(zhí)行超時時間，message具體消息內(nèi)容。?

job pool ：用來存放Job 的原信息，是個 map結構

Delay Bucket ：一組以時間為維度的有序隊列（這里只存放 job Id），bucket的數(shù)據(jù)結構就是redis的zset，將其分為多個bucket是為了提高掃描速度，降低消息延遲

Timer： 實時掃描各個 Bucket，并將delay時間小于等于當前時間的job放入到對應的 Ready Queue。

*?自己實現(xiàn)中，此處的Ready Queue?替換一個共同的kafka topic出口：存放處于Ready狀態(tài)的Job，以供客戶端消費程序消費。timer 到時間直接發(fā)送到 kafka ?

對比 Redisson 改動點

• 去除原有redisson 延時隊列 sub/pub實現(xiàn)timer思路，采用輪詢 zset 頭部節(jié)點，判斷是否已到過期時間進行判斷。

• 加入線程池概念，加快消息處理，減少延時消息時間誤差。

• cluster 模式，可用 redis 的 setnx命令實現(xiàn)簡單的分布式鎖，以保證集群中每次只有一個timer thread執(zhí)行。

個人改動點

• 做成通用性服務，提供統(tǒng)一的push topic，和統(tǒng)一的pull topic

整體執(zhí)行流程：

• 各個業(yè)務方把任務發(fā)給入口topic，生成延遲任務，放入某個桶

• 定時器時刻輪詢各個桶，當時間到達，發(fā)送消息任務到Kafka

• 消費端可以從?Kafka?共同出口中取到任務，做相應的業(yè)務邏輯

• 出口topic接收到消息，Kafka確認應答一次，保證消息不丟失?

微服務延時隊列整體架構圖

例子：

kafka 共同入口 delay_entrance_topic 格式：

kafka 共同出口 delay_exit_topic 格式：

• 減少延時時間誤差，使用線程池加快輪訓判斷時間到期
• cluster模式，防止其中一臺服務器掛了無法使用，高可用設計，使用定時器維護路由
• cluseter模式中，timer 代碼邏輯需要設置分布式鎖，防止多臺服務器同時執(zhí)行
• 消息可靠性：保證至少被消費一次，消費不成功，未應答，會重新投遞一次。

消息持久化問題：基于Redis自身的持久化特性，如果Redis數(shù)據(jù)丟失，意味著延遲消息的丟失，不過可以做主備和集群保證。這個可以考慮后續(xù)優(yōu)化將消息持久化到MangoDB中。

Netty 時間輪

HashedWheelTimer 流程圖

tickDuration: 每個格子的時間大小，每次轉(zhuǎn)動的時間

ticksPerWheel：時間輪數(shù)組大小

HashedWheelBucket:數(shù)組，記錄 header,tail

HashedWheelTimeOut: 延時任務載體，放于Bucket 數(shù)組中，屬性有：前后指針，round 數(shù)等

如果把時間輪看作一個map，那么 tickPerWheel 就為map的size，時間輪開始的時候，會設置一個 startTime，即每ticket都可算出延時時間，也就是 map 的key，value 為bucket。

核心代碼，線程 for循環(huán)，校驗 此刻的 bucket的鏈表是否到了執(zhí)行時間，到了就立即執(zhí)行，且 ticket+1，往下走。沒有則會sleep一會兒。

????????????long?deadline?=?tickDuration?*?(tick?+?1);
            for (;;) {                // 相對時間                final long currentTime = System.nanoTime() - startTime;
                long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
                // <=0 說明可以撥動時鐘了                if (sleepTimeMs <= 0) {                    if (currentTime == Long.MIN_VALUE) {                        return -Long.MAX_VALUE;                    } else {                        return currentTime;                    }                }

                // 這里是為了兼容 Windows 平臺                if (PlatformDependent.isWindows()) {                    sleepTimeMs = sleepTimeMs / 10 * 10;                }
                try {                    Thread.sleep(sleepTimeMs);                } catch (InterruptedException ignored) {                    if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {                        return Long.MIN_VALUE;                    }                }            }????????}

kafka 時間輪

在普通時間輪的基礎上，以空間換時間的思路，用 DelayQueue 去存儲每個 Bucket，DelayQueue 內(nèi)部有個 PriorityQueue，以每個bucket的延時時間進行大小排序，隊首的bucket就為將要執(zhí)行的任務，如果到期了，則可以直接取出執(zhí)行，未到則阻塞。依次循環(huán)取空優(yōu)先隊列。

其中的比對時間到期，交給底層api去做，Condition.awaitNanos() -> parkNanos() 核心代碼：

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) {        //驅(qū)動時間輪        timingWheel.advanceClock(bucket.getExpiration())       //循環(huán)buckek也就是任務列表，任務列表一個個繼續(xù)添加進時間輪以此來升級或者降級時間輪，把過期任務找出來執(zhí)行        bucket.flush(reinsert)        //這里就是從延遲隊列取出bucket，bucket是有延遲時間的，取出代表該bucket過期，通過bucket能取到bucket包含的任務列表        bucket = delayQueue.poll()      }    } finally {      writeLock.unlock()    }    true  } else {    false  }}

小問題

對于時間計算方面的問題，底層系統(tǒng)提供的api為什么效率更低呢？

它應該也是循環(huán)檢查到期時間，看到有的同學說，更推薦使用底層api，原理是一樣的，它為什么就比放在外面要好些呢？如果有知道的同學，也可在評論區(qū)告訴作者，感恩！

XXL_JOB

主要有兩個線程：scheduleThread 負責把 5s 之后要執(zhí)行的任務，從 db 中掃出來，放到 時間輪 容器中。

ringThread 負責把時針指向的每個到期的任務鏈表，交由快慢線程，rpc調(diào)用指給調(diào)度器執(zhí)行。

分布式任務調(diào)度，多個執(zhí)行器。任務持久化，任務統(tǒng)一先入庫，延時也是用的傳統(tǒng)時間輪。?

兩個非常核心的問題：

• 一定先給所有的延時任務排序

• 比對時間問題，到了任務執(zhí)行時間取出來

?所以，如果想自己設計一個延時隊列，關鍵是確定這兩個核心問題怎么解決，其余的根據(jù)自己的業(yè)務場景進行調(diào)整吧。

• https://juejin.cn/post/6845166891225317384

• https://juejin.cn/post/6910068006244581390
• https://juejin.cn/post/6976412313981026318