延時(shí)消息常見實(shí)現(xiàn)方案

延時(shí)消息（定時(shí)消息）指的在分布式異步消息場(chǎng)景下，生產(chǎn)端發(fā)送一條消息，希望在指定延時(shí)或者指定時(shí)間點(diǎn)被消費(fèi)端消費(fèi)到，而不是立刻被消費(fèi)。

延時(shí)消息適用的業(yè)務(wù)場(chǎng)景非常的廣泛，在分布式系統(tǒng)環(huán)境下，延時(shí)消息的功能一般會(huì)在下沉到中間件層，通常是 MQ 中內(nèi)置這個(gè)功能或者內(nèi)聚成一個(gè)公共基礎(chǔ)服務(wù)。

本文旨在探討常見延時(shí)消息的實(shí)現(xiàn)方案以及方案設(shè)計(jì)的優(yōu)缺點(diǎn)。

1. 基于外部存儲(chǔ)實(shí)現(xiàn)的方案

這里討論的外部存儲(chǔ)指的是在 MQ 本身自帶的存儲(chǔ)以外又引入的其他的存儲(chǔ)系統(tǒng)。

基于外部存儲(chǔ)的方案本質(zhì)上都是一個(gè)套路，將 MQ 和 延時(shí)模塊 區(qū)分開來，延時(shí)消息模塊是一個(gè)獨(dú)立的服務(wù)/進(jìn)程。延時(shí)消息先保留到其他存儲(chǔ)介質(zhì)中，然后在消息到期時(shí)再投遞到 MQ。當(dāng)然還有一些細(xì)節(jié)性的設(shè)計(jì)，比如消息進(jìn)入的延時(shí)消息模塊時(shí)已經(jīng)到期則直接投遞這類的邏輯，這里不展開討論。

下述方案不同的是，采用了不同的存儲(chǔ)系統(tǒng)。

基于 數(shù)據(jù)庫（如MySQL）

基于關(guān)系型數(shù)據(jù)庫（如MySQL）延時(shí)消息表的方式來實(shí)現(xiàn)。

CREATE TABLE `delay_msg` (  `id` bigint unsigned NOT NULL AUTO_INCREMENT,  `delivery_time` DATETIME NOT NULL COMMENT '投遞時(shí)間',  `payloads` blob COMMENT '消息內(nèi)容',  PRIMARY KEY (`id`),  KEY `time_index` (`delivery_time`))

通過定時(shí)線程定時(shí)掃描到期的消息，然后進(jìn)行投遞。定時(shí)線程的掃描間隔理論上就是你延時(shí)消息的最小時(shí)間精度。

優(yōu)點(diǎn)：

• 實(shí)現(xiàn)簡(jiǎn)單；

缺點(diǎn)：

• B+Tree索引不適合消息場(chǎng)景的大量寫入；

RocksDB 的方案其實(shí)就是在上述方案上選擇了比較合適的存儲(chǔ)介質(zhì)。

RocksDB 在筆者之前的文章中有聊過，LSM 樹更適合大量寫入的場(chǎng)景。滴滴開源的DDMQ中的延時(shí)消息模塊 Chronos 就是采用了這個(gè)方案。

DDMQ 這個(gè)項(xiàng)目簡(jiǎn)單來說就是在 RocketMQ 外面加了一層統(tǒng)一的代理層，在這個(gè)代理層就可以做一些功能維度的擴(kuò)展。延時(shí)消息的邏輯就是代理層實(shí)現(xiàn)了對(duì)延時(shí)消息的轉(zhuǎn)發(fā)，如果是延時(shí)消息，會(huì)先投遞到 RocketMQ 中 Chronos 專用的 topic 中。延時(shí)消息模塊 Chronos 消費(fèi)得到延時(shí)消息轉(zhuǎn)儲(chǔ)到 RocksDB，后面就是類似的邏輯了，定時(shí)掃描到期的消息，然后往 RocketMQ 中投遞。

這個(gè)方案老實(shí)說是一個(gè)比較重的方案。因?yàn)榛?RocksDB 來實(shí)現(xiàn)的話，從數(shù)據(jù)可用性的角度考慮，你還需要自己去處理多副本的數(shù)據(jù)同步等邏輯。

優(yōu)點(diǎn)：

• RocksDB LSM 樹很適合消息場(chǎng)景的大量寫入；

缺點(diǎn)：

• 實(shí)現(xiàn)方案較重，如果你采用這個(gè)方案，需要自己實(shí)現(xiàn) RocksDB 的數(shù)據(jù)容災(zāi)邏輯；

再來聊聊 Redis 的方案。下面放一個(gè)比較完善的方案。

本方案來源于：基于Redis實(shí)現(xiàn)延時(shí)隊(duì)列服務(wù)

• Messages Pool 所有的延時(shí)消息存放，結(jié)構(gòu)為KV結(jié)構(gòu)，key為消息ID，value為一個(gè)具體的message（這里選擇Redis Hash結(jié)構(gòu)主要是因?yàn)閔ash結(jié)構(gòu)能存儲(chǔ)較大的數(shù)據(jù)量，數(shù)據(jù)較多時(shí)候會(huì)進(jìn)行漸進(jìn)式rehash擴(kuò)容，并且對(duì)于HSET和HGET命令來說時(shí)間復(fù)雜度都是O(1)）

• Delayed Queue是16個(gè)有序隊(duì)列（隊(duì)列支持水平擴(kuò)展），結(jié)構(gòu)為ZSET，value 為 messages pool中消息ID，score為過期時(shí)間**（分為多個(gè)隊(duì)列是為了提高掃描的速度）**

• Worker 代表處理線程，通過定時(shí)任務(wù)掃描 Delayed Queue 中到期的消息

這個(gè)方案選用 Redis 存儲(chǔ)在我看來有幾點(diǎn)考慮。

• Redis ZSET 很適合實(shí)現(xiàn)延時(shí)隊(duì)列

• 性能問題，雖然 ZSET 插入是一個(gè) O(logn) 的操作，但是Redis 基于內(nèi)存操作，并且內(nèi)部做了很多性能方面的優(yōu)化。

但是這個(gè)方案其實(shí)也有需要斟酌的地方，上述方案通過創(chuàng)建多個(gè) Delayed Queue 來滿足對(duì)于并發(fā)性能的要求，但這也帶來了多個(gè) Delayed Queue 如何在多個(gè)節(jié)點(diǎn)情況下均勻分配，并且很可能出現(xiàn)到期消息并發(fā)重復(fù)處理的情況，是否要引入分布式鎖之類的并發(fā)控制設(shè)計(jì)？

在量不大的場(chǎng)景下，上述方案的架構(gòu)其實(shí)可以蛻化成主從架構(gòu)，只允許主節(jié)點(diǎn)來處理任務(wù)，從節(jié)點(diǎn)只做容災(zāi)備份。實(shí)現(xiàn)難度更低更可控。

上述幾個(gè)方案中，都通過線程定時(shí)掃描的方案來獲取到期的消息。

定時(shí)線程的方案在消息量較少的時(shí)候，會(huì)浪費(fèi)資源，在消息量非常多的時(shí)候，又會(huì)出現(xiàn)因?yàn)閽呙栝g隔設(shè)置不合理導(dǎo)致延時(shí)時(shí)間不準(zhǔn)確的問題?？梢越柚?JDK Timer 類中的思想，通過 wait-notify 來節(jié)省 CPU 資源。

獲取中最近的延時(shí)消息，然后wait(執(zhí)行時(shí)間-當(dāng)前時(shí)間)，這樣就不需要浪費(fèi)資源到達(dá)時(shí)間時(shí)會(huì)自動(dòng)響應(yīng)，如果有新的消息進(jìn)入，并且比我們等待的消息還要小，那么直接notify喚醒，重新獲取這個(gè)更小的消息，然后又wait，如此循環(huán)。

2. 開源 MQ 中的實(shí)現(xiàn)方案

再來講講目前自帶延時(shí)消息功能的開源MQ，它們是如何實(shí)現(xiàn)的

• Level 數(shù)固定，每個(gè) Level 有自己的定時(shí)器，開銷不大
• 將 Level 相同的消息放入到同一個(gè) Queue 中，保證了同一 Level 消息的順序性；不同 Level 放到不同的 Queue 中，保證了投遞的時(shí)間準(zhǔn)確性；
• 通過只支持固定的Level，將不同延時(shí)消息的排序變成了固定Level Topic 的追加寫操作

• Level 配置的修改代價(jià)太大，固定 Level 不靈活
• CommitLog 會(huì)因?yàn)檠訒r(shí)消息的存在變得很大

• 內(nèi)存開銷：?維護(hù)延時(shí)消息索引的隊(duì)列是放在堆外內(nèi)存中的，并且這個(gè)隊(duì)列是以訂閱組（Kafka中的消費(fèi)組）為維度的，比如你這個(gè) Topic 有 N 個(gè)訂閱組，那么如果你這個(gè) Topic 使用了延時(shí)消息，就會(huì)創(chuàng)建 N 個(gè) 隊(duì)列；并且隨著延時(shí)消息的增多，時(shí)間跨度的增加，每個(gè)隊(duì)列的內(nèi)存占用也會(huì)上升。（是的，在這個(gè)方案下，支持任意的延時(shí)消息反而有可能讓這個(gè)缺陷更嚴(yán)重）
• 故障轉(zhuǎn)移之后延時(shí)消息索引隊(duì)列的重建時(shí)間開銷：?對(duì)于跨度時(shí)間長(zhǎng)的大規(guī)模延時(shí)消息，重建時(shí)間可能會(huì)到小時(shí)級(jí)別。（摘自 Pulsar 官方公眾號(hào)文章）
• 存儲(chǔ)開銷：延時(shí)消息的時(shí)間跨度會(huì)影響到 Pulsar 中已經(jīng)消費(fèi)的消息數(shù)據(jù)的空間回收。打個(gè)比方，你的 Topic 如果業(yè)務(wù)上要求支持一個(gè)月跨度的延時(shí)消息，然后你發(fā)了一個(gè)延時(shí)一個(gè)月的消息，那么你這個(gè) Topic 中底層的存儲(chǔ)就會(huì)保留整整一個(gè)月的消息數(shù)據(jù)，即使這一個(gè)月中99%的正常消息都已經(jīng)消費(fèi)了。

如果對(duì)時(shí)間輪不熟悉的可以閱讀筆者的這篇文章?從 Kafka 看時(shí)間輪算法設(shè)計(jì)

• 時(shí)間輪算法適合延時(shí)/定時(shí)消息的場(chǎng)景，省去延時(shí)消息的排序，插入刪除操作都是 O(1) 的時(shí)間復(fù)雜度；
• 通過多級(jí)時(shí)間輪設(shè)計(jì)，支持了超大時(shí)間跨度的延時(shí)消息；
• 通過延時(shí)加載，內(nèi)存中只會(huì)有最近要消費(fèi)的消息，更久的延時(shí)消息會(huì)被存儲(chǔ)在磁盤中，對(duì)內(nèi)存友好；
• 延時(shí)消息單獨(dú)存儲(chǔ)（schedule log），不會(huì)影響到正常消息的空間回收；

作者：Richard_Yi

來源：juejin.cn/post/7052894117105238053