延遲消息是實(shí)際開(kāi)發(fā)中一個(gè)非常有用的功能，本文第一部分從整體上介紹秒級(jí)精度延遲消息的實(shí)現(xiàn)思路，在第二部分結(jié)合RocketMQ的延遲消息實(shí)現(xiàn)，進(jìn)行細(xì)致的講解颅停，點(diǎn)出關(guān)鍵部分的源碼。第三步介紹延遲消息與消息重試的關(guān)系掠拳。

1 延遲消息介紹

基本概念：延遲消息是指生產(chǎn)者發(fā)送消息發(fā)送消息后癞揉，不能立刻被消費(fèi)者消費(fèi)，需要等待指定的時(shí)間后才可以被消費(fèi)。

場(chǎng)景案例：用戶(hù)下了一個(gè)訂單之后喊熟，需要在指定時(shí)間內(nèi)(例如30分鐘)進(jìn)行支付柏肪，在到期之前可以發(fā)送一個(gè)消息提醒用戶(hù)進(jìn)行支付。

一些消息中間件的Broker端內(nèi)置了延遲消息支持的能力逊移，如：

NSQ：這是一個(gè)go語(yǔ)言的消息中間件预吆，其通過(guò)內(nèi)存中的優(yōu)先級(jí)隊(duì)列來(lái)保存延遲消息，支持秒級(jí)精度胳泉，最多2個(gè)小時(shí)延遲。Java中也有對(duì)應(yīng)的實(shí)現(xiàn)岩遗，如ScheduledThreadPoolExecutor內(nèi)部實(shí)際上也是使用了優(yōu)先級(jí)隊(duì)列扇商。

QMQ：采用雙重時(shí)間輪實(shí)現(xiàn)。https://www.toutiao.com/i6851807550690722312/

RabbitMQ：需要安裝一個(gè)rabbitmq_delayed_message_exchange插件宿礁。

RocketMQ：RocketMQ 開(kāi)源版本延遲消息臨時(shí)存儲(chǔ)在一個(gè)內(nèi)部主題SCHEDULE_TOPIC_XXXX中案铺， ? ?不支持任意時(shí)間精度，支持特定的 level梆靖，例如定時(shí) 5s控汉，10s，1m 等返吻。

Broker端內(nèi)置延遲消息處理能力姑子，核心實(shí)現(xiàn)思路都是一樣：將延遲消息通過(guò)一個(gè)臨時(shí)存儲(chǔ)進(jìn)行暫存，到期后才投遞到目標(biāo)Topic中测僵。如下圖所示：

?

步驟說(shuō)明如下：

1. producer要將一個(gè)延遲消息發(fā)送到某個(gè)Topic中

2. Broker判斷這是一個(gè)延遲消息后街佑，將其通過(guò)臨時(shí)存儲(chǔ)進(jìn)行暫存。

3. Broker內(nèi)部通過(guò)一個(gè)延遲服務(wù)(delay service)檢查消息是否到期捍靠，將到期的消息投遞到目標(biāo)Topic中沐旨。這個(gè)的延遲服務(wù)名字為delay service，不同消息中間件的延遲服務(wù)模塊名稱(chēng)可能不同榨婆。

4. 消費(fèi)者消費(fèi)目標(biāo)topic中的延遲投遞的消息

顯然磁携，臨時(shí)存儲(chǔ)模塊和延遲服務(wù)模塊，是延遲消息實(shí)現(xiàn)的關(guān)鍵良风。上圖中谊迄，臨時(shí)存儲(chǔ)和延遲服務(wù)都是在Broker內(nèi)部實(shí)現(xiàn)，對(duì)業(yè)務(wù)透明拖吼。

此外鳞上， 還有一些消息中間件原生并不支持延遲消息，如Kafka吊档。在這種情況下篙议，可以選擇對(duì)Kafka進(jìn)行改造，但是成本較大。另外一種方式是使用第三方臨時(shí)存儲(chǔ)鬼贱，并加一層代理移怯。

第三方存儲(chǔ)選型要求：

對(duì)于第三方臨時(shí)存儲(chǔ)，其需要滿(mǎn)足以下幾個(gè)特點(diǎn)：

高性能：

????寫(xiě)入延遲要低这难，MQ的一個(gè)重要作用是削峰填谷舟误，在選擇臨時(shí)存儲(chǔ)時(shí)，寫(xiě)入性能必須要高姻乓，關(guān)系型數(shù)據(jù)庫(kù)(如Mysql)通常不滿(mǎn)足需求嵌溢。

高可靠：

????延遲消息寫(xiě)入后，不能丟失蹋岩，需要進(jìn)行持久化赖草，并進(jìn)行備份

支持排序：

????支持按照某個(gè)字段對(duì)消息進(jìn)行排序，對(duì)于延遲消息需要按照時(shí)間進(jìn)行排序剪个。普通消息通常先發(fā)送的會(huì)被先消費(fèi)秧骑，延遲消息與普通消息不同，需要進(jìn)行排序扣囊。例如先發(fā)一條延遲10s的消息乎折，再發(fā)一條延遲5s的消息，那么后發(fā)送的消息需要被先消費(fèi)侵歇。

支持長(zhǎng)時(shí)間保存：

????一些業(yè)務(wù)的延遲消息骂澄，需要延遲幾個(gè)月，甚至更長(zhǎng)盒至，所以延遲消息必須能長(zhǎng)時(shí)間保留酗洒。不過(guò)通常不建議延遲太長(zhǎng)時(shí)間，存儲(chǔ)成本比較大枷遂，且業(yè)務(wù)邏輯可能已經(jīng)發(fā)生變化樱衷，已經(jīng)不需要消費(fèi)這些消息。

例如酒唉，滴滴開(kāi)源的消息中間件DDMQ矩桂，底層消息中間件的基礎(chǔ)上加了一層代理，獨(dú)立部署延遲服務(wù)模塊痪伦，使用rocksdb進(jìn)行臨時(shí)存儲(chǔ)侄榴。rocksdb是一個(gè)高性能的KV存儲(chǔ)，并支持排序网沾。

此時(shí)對(duì)于延遲消息的流轉(zhuǎn)如下圖所示：

?

說(shuō)明如下：

1 生產(chǎn)者將發(fā)送給producer proxy癞蚕，proxy判斷是延遲消息，將其投遞到一個(gè)緩沖Topic中辉哥；

2 delay service啟動(dòng)消費(fèi)者桦山，用于從緩沖topic中消費(fèi)延遲消息攒射，以時(shí)間為key，存儲(chǔ)到rocksdb中恒水；

3 delay service判斷消息到期后会放，將其投遞到目標(biāo)Topic中。

4 消費(fèi)者消費(fèi)目標(biāo)topic中的數(shù)據(jù)

這種方式的好處是钉凌，因?yàn)閐elay service的延遲投遞能力是獨(dú)立于broker實(shí)現(xiàn)的咧最，不需要對(duì)broker做任何改造，對(duì)于任意MQ類(lèi)型都可以提供支持延遲消息的能力御雕，例如DDMQ對(duì)RocketMQ矢沿、Kafka都提供了秒級(jí)精度的延遲消息投遞能力，但是Kafka本身并不支持延遲消息饮笛，而RocketMQ雖然支持延遲消息咨察，但不支持秒級(jí)精度。

事實(shí)上福青，DDMQ還提供了很多其他功能，僅僅從延遲消息的角度脓诡，完全沒(méi)有必要使用這個(gè)proxy无午，直接將消息投遞到緩沖Topic中，之后通過(guò)delay service完成延遲投遞邏輯即可祝谚。

具體到delay service模塊的實(shí)現(xiàn)上宪迟，也有一些重要的細(xì)節(jié)：

1 為了保證服務(wù)的高可用，delay service也是需要部署多個(gè)節(jié)點(diǎn)交惯。

2 為了保證數(shù)據(jù)不丟失次泽，每個(gè)delay service節(jié)點(diǎn)都需要消費(fèi)緩沖Topic中的全量數(shù)據(jù)，保存到各自的持久化存儲(chǔ)中席爽，這樣就有了多個(gè)備份意荤，并需要以時(shí)間為key。不過(guò)因?yàn)槭歉髯岳≈欢停⒉荒鼙ＷC強(qiáng)一致玖像。如果一定要強(qiáng)一致，那么delay service就不需要內(nèi)置存儲(chǔ)實(shí)現(xiàn)齐饮，可以借助于其他支持強(qiáng)一致的存儲(chǔ)捐寥。

3 為了避免重復(fù)投遞，delay service需要進(jìn)行選主祖驱，可以借助于zookeeper握恳、etcd等實(shí)現(xiàn)。只有master可以通過(guò)生產(chǎn)者投遞到目標(biāo)Topic中捺僻，其他節(jié)點(diǎn)處于備用狀態(tài)乡洼。否則，如果每個(gè)節(jié)點(diǎn)進(jìn)行都投遞，那么延遲消息就會(huì)被投遞多次就珠，造成消費(fèi)重復(fù)寇壳。

4 master要記錄自己當(dāng)前投遞到的時(shí)間到一個(gè)共享存儲(chǔ)中，如果master掛了妻怎，從slave節(jié)點(diǎn)中選出一個(gè)新的master節(jié)點(diǎn)壳炎，從之前記錄時(shí)間繼續(xù)開(kāi)始投遞。

5 延遲消息的取消：一些延遲消息在未到期之前逼侦，可能希望進(jìn)行取消匿辩。通常取消邏輯實(shí)現(xiàn)較為復(fù)雜，且不夠精確榛丢。對(duì)于那些已經(jīng)快要到期的消息铲球，可能還未取消之前，已經(jīng)發(fā)送出去了晰赞，因此需要在消費(fèi)者端做檢查稼病，才能萬(wàn)無(wú)一失。

2 RocketMQ中的延遲消息

開(kāi)源RocketMQ支持延遲消息掖鱼，但是不支持秒級(jí)精度然走。默認(rèn)支持18個(gè)level的延遲消息，這是通過(guò)broker端的messageDelayLevel配置項(xiàng)確定的戏挡，如下：

messageDelayLevel=1s?5s?10s?30s?1m?2m?3m?4m?5m?6m?7m?8m?9m?10m?20m?30m?1h?2h

Broker在啟動(dòng)時(shí)芍瑞，內(nèi)部會(huì)創(chuàng)建一個(gè)內(nèi)部主題：SCHEDULE_TOPIC_XXXX，根據(jù)延遲level的個(gè)數(shù)褐墅，創(chuàng)建對(duì)應(yīng)數(shù)量的隊(duì)列拆檬，也就是說(shuō)18個(gè)level對(duì)應(yīng)了18個(gè)隊(duì)列。注意妥凳，這并不是說(shuō)這個(gè)內(nèi)部主題只會(huì)有18個(gè)隊(duì)列竟贯，因?yàn)锽roker通常是集群模式部署的，因此每個(gè)節(jié)點(diǎn)都有18個(gè)隊(duì)列猾封。

延遲級(jí)別的值可以進(jìn)行修改澄耍，以滿(mǎn)足自己的業(yè)務(wù)需求，可以修改/添加新的level晌缘。例如：你想支持2天的延遲齐莲，修改最后一個(gè)level的值為2d，這個(gè)時(shí)候依然是18個(gè)level磷箕；也可以增加一個(gè)2d选酗，這個(gè)時(shí)候總共就有19個(gè)level。

可以看到這里并不支持秒級(jí)精度岳枷，按照《rocketmq developer guide》中的說(shuō)法芒填，是為了避免在broker對(duì)消息進(jìn)行排序呜叫，造成性能影響。不過(guò)筆者考慮殿衰，之所以不支持朱庆，更多應(yīng)該是商業(yè)上的考慮。

生產(chǎn)者發(fā)送延遲消息：

生產(chǎn)者在發(fā)送延遲消息非常簡(jiǎn)單闷祥，只需要設(shè)置一個(gè)延遲級(jí)別即可娱颊，注意不是具體的延遲時(shí)間，如：

Message?msg=new?Message();

msg.setTopic("TopicA");

msg.setTags("Tag");

msg.setBody("this?is?a?delay?message".getBytes());

//設(shè)置延遲level為5凯砍，對(duì)應(yīng)延遲1分鐘

msg.setDelayTimeLevel(5);

producer.send(msg);

如果設(shè)置的延遲level超過(guò)最大值箱硕，那么將會(huì)重置最最大值。

Broker端延遲消息處理：

延遲消息在RocketMQ Broker端的流轉(zhuǎn)如下圖所示：

?

可以看到悟衩，總共有6個(gè)步驟剧罩，下面會(huì)對(duì)這6個(gè)步驟進(jìn)行詳細(xì)的講解：

1 修改消息Topic名稱(chēng)和隊(duì)列信息

2 轉(zhuǎn)發(fā)消息到延遲主題的CosumeQueue中

3 延遲服務(wù)消費(fèi)SCHEDULE_TOPIC_XXXX消息

4 將信息重新存儲(chǔ)到CommitLog中

5 將消息投遞到目標(biāo)Topic中

6 消費(fèi)者消費(fèi)目標(biāo)topic中的數(shù)據(jù)

第一步：修改消息Topic名稱(chēng)和隊(duì)列信息

RocketMQ Broker端在存儲(chǔ)生產(chǎn)者寫(xiě)入的消息時(shí)，首先都會(huì)將其寫(xiě)入到CommitLog中座泳。之后根據(jù)消息中的Topic信息和隊(duì)列信息惠昔，將其轉(zhuǎn)發(fā)到目標(biāo)Topic的指定隊(duì)列(ConsumeQueue)中。

由于消息一旦存儲(chǔ)到ConsumeQueue中挑势，消費(fèi)者就能消費(fèi)到舰罚，而延遲消息不能被立即消費(fèi)，所以這里將Topic的名稱(chēng)修改為SCHEDULE_TOPIC_XXXX薛耻，并根據(jù)延遲級(jí)別確定要投遞到哪個(gè)隊(duì)列下。

同時(shí)赏陵，還會(huì)將消息原來(lái)要發(fā)送到的目標(biāo)Topic和隊(duì)列信息存儲(chǔ)到消息的屬性中饼齿。相關(guān)源碼如下所示：

org.apache.rocketmq.store.CommitLog#putMessage

public?PutMessageResult?putMessage(final?MessageExtBrokerInner?msg)?{

...

????//?如果是延遲消息

????if?(msg.getDelayTimeLevel()?>?0)?{

????????//如果設(shè)置的級(jí)別超過(guò)了最大級(jí)別，重置延遲級(jí)別

????????if?(msg.getDelayTimeLevel()?>?this.defaultMessageStore.getScheduleMessageService()

???????????????????????????????????????????????????????????????????.getMaxDelayLevel())?{

????????????msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService()

????????????????????????????????????????????????????????????????????.getMaxDelayLevel());

????????}

????????//修改Topic的投遞目標(biāo)為內(nèi)部主題SCHEDULE_TOPIC_XXXX

????????topic?=?ScheduleMessageService.SCHEDULE_TOPIC;

????????//根據(jù)delayLevel蝙搔，確定將消息投遞到SCHEDULE_TOPIC_XXXX內(nèi)部的哪個(gè)隊(duì)列中

????????queueId?=?ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

????????//?記錄原始topic,?queueId

????????MessageAccessor.putProperty(msg,?MessageConst.PROPERTY_REAL_TOPIC,?msg.getTopic());

????????MessageAccessor.putProperty(msg,?MessageConst.PROPERTY_REAL_QUEUE_ID,?

????????????????????????????????????String.valueOf(msg.getQueueId()));

????????msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

????????//更新消息投遞目標(biāo)為SCHEDULE_TOPIC_XXXX和queueId

????????msg.setTopic(topic);

????????msg.setQueueId(queueId);

????}

...

第二步：轉(zhuǎn)發(fā)消息到延遲主題的CosumeQueue中

CommitLog中的消息轉(zhuǎn)發(fā)到CosumeQueue中是異步進(jìn)行的缕溉。在轉(zhuǎn)發(fā)過(guò)程中，會(huì)對(duì)延遲消息進(jìn)行特殊處理吃型，主要是計(jì)算這條延遲消息需要在什么時(shí)候進(jìn)行投遞证鸥。

投遞時(shí)間=消息存儲(chǔ)時(shí)間(storeTimestamp)?+?延遲級(jí)別對(duì)應(yīng)的時(shí)間

需要注意的是，會(huì)將計(jì)算出的投遞時(shí)間當(dāng)做消息Tag的哈希值存儲(chǔ)到CosumeQueue中勤晚，CosumeQueue單個(gè)存儲(chǔ)單元組成結(jié)構(gòu)如下圖所示：

?

其中：

Commit Log Offset：記錄在CommitLog中的位置枉层。

Size：記錄消息的大小

Message Tag HashCode：記錄消息Tag的哈希值，用于消息過(guò)濾赐写。特別的鸟蜡，對(duì)于延遲消息，這個(gè)字段記錄的是消息的投遞時(shí)間戳挺邀。這也是為什么java中hashCode方法返回一個(gè)int型揉忘，只占用4個(gè)字節(jié)跳座，而這里Message Tag HashCode字段確設(shè)計(jì)成8個(gè)字節(jié)的原因。

相關(guān)源碼參見(jiàn)：

CommitLog#checkMessageAndReturnSize

public?DispatchRequest?checkMessageAndReturnSize(java.nio.ByteBuffer?byteBuffer,?final?boolean?checkCRC,

????final?boolean?readBody)?{

??...

??//?Timing?message?processing

??{

??????//如果消息需要投遞到延遲主題SCHEDULE_TOPIC_XXX中

??????String?t?=?propertiesMap.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);

??????if?(ScheduleMessageService.SCHEDULE_TOPIC.equals(topic)?&&?t?!=?null)?{

??????????int?delayLevel?=?Integer.parseInt(t);

??????????if?(delayLevel?>?this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel())?{

??????????????delayLevel?=?this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel();

??????????}

??????????//如果延遲級(jí)別大于0泣矛，計(jì)算目標(biāo)投遞時(shí)間疲眷，并將其當(dāng)做tag哈希值??

??????????if?(delayLevel?>?0)?{

??????????????tagsCode?=?this.defaultMessageStore.getScheduleMessageService()

????????????????????????????.computeDeliverTimestamp(delayLevel,storeTimestamp);

??????????}

??????}

??}

??...

第三步：延遲服務(wù)消費(fèi)SCHEDULE_TOPIC_XXXX消息

Broker內(nèi)部有一個(gè)ScheduleMessageService類(lèi)，其充當(dāng)延遲服務(wù)您朽，消費(fèi)SCHEDULE_TOPIC_XXXX中的消息狂丝，并投遞到目標(biāo)Topic中。

ScheduleMessageService在啟動(dòng)時(shí)虚倒，其會(huì)創(chuàng)建一個(gè)定時(shí)器Timer美侦，并根據(jù)延遲級(jí)別的個(gè)數(shù)，啟動(dòng)對(duì)應(yīng)數(shù)量的TimerTask魂奥，每個(gè)TimerTask負(fù)責(zé)一個(gè)延遲級(jí)別的消費(fèi)與投遞菠剩。

相關(guān)源碼如下所示：

ScheduleMessageService#start

public?void?start()?{

????if?(started.compareAndSet(false,?true))?{

????????//1?創(chuàng)建定時(shí)器Timer

????????this.timer?=?new?Timer("ScheduleMessageTimerThread",?true);

????????//2?針對(duì)每個(gè)延遲級(jí)別，創(chuàng)建一個(gè)TimerTask????

????????//2.1?迭代每個(gè)延遲級(jí)別：delayLevelTable是一個(gè)Map記錄了每個(gè)延遲級(jí)別對(duì)應(yīng)的延遲時(shí)間

????????for?(Map.Entry<Integer,?Long>?entry?:?this.delayLevelTable.entrySet())?{

????????????//2.2?獲得每個(gè)每個(gè)延遲級(jí)別的level和對(duì)應(yīng)的延遲時(shí)間

????????????Integer?level?=?entry.getKey();

????????????Long?timeDelay?=?entry.getValue();

????????????Long?offset?=?this.offsetTable.get(level);

????????????if?(null?==?offset)?{

????????????????offset?=?0L;

????????????}

????????????//2.3?針對(duì)每個(gè)級(jí)別創(chuàng)建一個(gè)對(duì)應(yīng)的TimerTask

????????????if?(timeDelay?!=?null)?{

????????????????this.timer.schedule(new?DeliverDelayedMessageTimerTask(level,?offset),?

????????????????????????????????????????????????????????????????????FIRST_DELAY_TIME);

????????????}

????????}

????...

需要注意的是耻煤，每個(gè)TimeTask在檢查消息是否到期時(shí)具壮，首先檢查對(duì)應(yīng)隊(duì)列中尚未投遞第一條消息，如果這條消息沒(méi)到期哈蝇，那么之后的消息都不會(huì)檢查棺妓。如果到期了，則進(jìn)行投遞炮赦，并檢查之后的消息是否到期怜跑。

第四步：將信息重新存儲(chǔ)到CommitLog中

在將消息到期后，需要投遞到目標(biāo)Topic吠勘。由于在第一步已經(jīng)記錄了原來(lái)的Topic和隊(duì)列信息性芬，因此這里重新設(shè)置，再存儲(chǔ)到CommitLog即可剧防。此外植锉，由于之前Message Tag HashCode字段存儲(chǔ)的是消息的投遞時(shí)間，這里需要重新計(jì)算tag的哈希值后再存儲(chǔ)峭拘。

源碼參見(jiàn)：DeliverDelayedMessageTimerTask的messageTimeup方法了罪。

第五步：將消息投遞到目標(biāo)Topic中

這一步與第二步類(lèi)似喘垂，不過(guò)由于消息的Topic名稱(chēng)已經(jīng)改為了目標(biāo)Topic庭呜。因此消息會(huì)直接投遞到目標(biāo)Topic的ConsumeQueue中懈贺，之后消費(fèi)者即消費(fèi)到這條消息。

3 延遲消息與消息重試的關(guān)系

RocketMQ提供了消息重試的能力宵凌，在并發(fā)模式消費(fèi)的情況鞋囊，如果消費(fèi)失敗，可以返回一個(gè)枚舉值RECONSUME_LATER瞎惫，那么消息之后將會(huì)進(jìn)行重試溜腐。如：

consumer.registerMessageListener(new?MessageListenerConcurrently()?{

???????@Override

???????public?ConsumeConcurrentlyStatus?consumeMessage(List<MessageExt>?msgs,

???????????????????????????????????????ConsumeConcurrentlyContext?context)?{

???????????//處理消息译株，失敗，返回RECONSUME_LATER挺益，進(jìn)行重試

???????????return?ConsumeConcurrentlyStatus.RECONSUME_LATER;

???????}

???});

重試默認(rèn)會(huì)進(jìn)行重試16次歉糜。使用過(guò)RocketMQ消息重試功能的用戶(hù)，可能看到過(guò)以下這張圖：

第幾次重試與上次重試的間隔時(shí)間第幾次重試與上次重試的間隔時(shí)間

110 秒97 分鐘

230 秒108 分鐘

31 分鐘119 分鐘

42 分鐘1210 分鐘

53 分鐘1320 分鐘

64 分鐘1430 分鐘

75 分鐘151 小時(shí)

86 分鐘162 小時(shí)

細(xì)心地的讀者發(fā)現(xiàn)了望众，消息重試的16個(gè)級(jí)別匪补，實(shí)際上是把延遲消息18個(gè)級(jí)別的前兩個(gè)level去掉了，事實(shí)上烂翰，RocketMQ的消息重試也是基于延遲消息來(lái)完成的夯缺。在消息消費(fèi)失敗的情況下，將其重新當(dāng)做延遲消息投遞回Broker甘耿。

在投遞回去時(shí)踊兜，會(huì)跳過(guò)前兩個(gè)level，因此只重試16次佳恬。當(dāng)然捏境，消息重試還有一些其他的設(shè)計(jì)邏輯，在之后的文章將會(huì)進(jìn)行分析毁葱。

免費(fèi)學(xué)習(xí)視頻歡迎關(guān)注云圖智聯(lián)：https://e.yuntuzhilian.com/