深入理解RocketMQ延遲消息

本文轉載自查看原文 2022-02-12 21:19 1723 RocketMQ

延遲消息是實際開發中一個非常有用的功能，本文第一部分從整體上介紹秒級精度延遲消息的實現思路，在第二部分結合RocketMQ的延遲消息實現，進行細致的講解，點出關鍵部分的源碼。第三步介紹延遲消息與消息重試的關系。

延遲消息介紹

基本概念：延遲消息是指生產者發送消息發送消息后，不能立刻被消費者消費，需要等待指定的時間后才可以被消費。
場景案例：用戶下了一個訂單之后，需要在指定時間內(例如30分鍾)進行支付，在到期之前可以發送一個消息提醒用戶進行支付。
一些消息中間件的Broker端內置了延遲消息支持的能力，如：

NSQ：這是一個go語言的消息中間件，其通過內存中的優先級隊列來保存延遲消息，支持秒級精度，最多2個小時延遲。Java中也有對應的實現，如ScheduledThreadPoolExecutor內部實際上也是使用了優先級隊列。
QMQ：采用雙重時間輪實現。可參考：任意時間延時消息原理講解：設計與實現
RabbitMQ：需要安裝一個rabbitmq_delayed_message_exchange插件。
RocketMQ：RocketMQ 開源版本延遲消息臨時存儲在一個內部主題中，不支持任意時間精度，支持特定的 level，例如定時 5s，10s，1m 等。

Broker端內置延遲消息處理能力，核心實現思路都是一樣：將延遲消息通過一個臨時存儲進行暫存，到期后才投遞到目標Topic中。如下圖所示：

步驟說明如下：

producer要將一個延遲消息發送到某個Topic中
Broker判斷這是一個延遲消息后，將其通過臨時存儲進行暫存。
Broker內部通過一個延遲服務(delay service)檢查消息是否到期，將到期的消息投遞到目標Topic中。這個的延遲服務名字為delay service，不同消息中間件的延遲服務模塊名稱可能不同。
消費者消費目標topic中的延遲投遞的消息

顯然，臨時存儲模塊和延遲服務模塊，是延遲消息實現的關鍵。上圖中，臨時存儲和延遲服務都是在Broker內部實現，對業務透明。

此外，還有一些消息中間件原生並不支持延遲消息，如Kafka。在這種情況下，可以選擇對Kafka進行改造，但是成本較大。另外一種方式是使用第三方臨時存儲，並加一層代理。

第三方存儲選型要求：
對於第三方臨時存儲，其需要滿足以下幾個特點：

高性能：寫入延遲要低，MQ的一個重要作用是削峰填谷，在選擇臨時存儲時，寫入性能必須要高，關系型數據庫(如Mysql)通常不滿足需求。
高可靠：延遲消息寫入后，不能丟失，需要進行持久化，並進行備份
支持排序：支持按照某個字段對消息進行排序，對於延遲消息需要按照時間進行排序。普通消息通常先發送的會被先消費，延遲消息與普通消息不同，需要進行排序。例如先發一條延遲10s的消息，再發一條延遲5s的消息，那么后發送的消息需要被先消費。
支持長時間保存：一些業務的延遲消息，需要延遲幾個月，甚至更長，所以延遲消息必須能長時間保留。不過通常不建議延遲太長時間，存儲成本比較大，且業務邏輯可能已經發生變化，已經不需要消費這些消息。

例如，滴滴開源的消息中間件DDMQ，底層消息中間件的基礎上加了一層代理，獨立部署延遲服務模塊，使用rocksdb進行臨時存儲。rocksdb是一個高性能的KV存儲，並支持排序。

此時對於延遲消息的流轉如下圖所示：

說明如下：

生產者將發送給producer proxy，proxy判斷是延遲消息，將其投遞到一個緩沖Topic中；
delay service啟動消費者，用於從緩沖topic中消費延遲消息，以時間為key，存儲到rocksdb中；
delay service判斷消息到期后，將其投遞到目標Topic中。
消費者消費目標topic中的數據

這種方式的好處是，因為delay service的延遲投遞能力是獨立於broker實現的，不需要對broker做任何改造，對於任意MQ類型都可以提供支持延遲消息的能力。例如DDMQ對RocketMQ、Kafka都提供了秒級精度的延遲消息投遞能力，但是Kafka本身並不支持延遲消息，而開源版本的 RocketMQ 只支持幾個指定的延遲級別，並不支持秒級精度的定時消息。

事實上，DDMQ還提供了很多其他功能，僅僅從延遲消息的角度，完全沒有必要使用這個proxy，直接將消息投遞到緩沖Topic中，之后通過delay service完成延遲投遞邏輯即可。

具體到delay service模塊的實現上，也有一些重要的細節：

為了保證服務的高可用，delay service也是需要部署多個節點。
為了保證數據不丟失，每個delay service節點都需要消費緩沖Topic中的全量數據，保存到各自的持久化存儲中，這樣就有了多個備份，並需要以時間為key。不過因為是各自拉取，並不能保證強一致。如果一定要強一致，那么delay service就不需要內置存儲實現，可以借助於其他支持強一致的存儲。
為了避免重復投遞，delay service需要進行選主，可以借助於zookeeper、etcd等實現。只有master可以通過生產者投遞到目標Topic中，其他節點處於備用狀態。否則，如果每個節點進行都投遞，那么延遲消息就會被投遞多次，造成消費重復。
master要記錄自己當前投遞到的時間到一個共享存儲中，如果master掛了，從slave節點中選出一個新的master節點，從之前記錄時間繼續開始投遞。
延遲消息的取消：一些延遲消息在未到期之前，可能希望進行取消。通常取消邏輯實現較為復雜，且不夠精確。對於那些已經快要到期的消息，可能還未取消之前，已經發送出去了，因此需要在消費者端做檢查，才能萬無一失。

RocketMQ中的延遲消息

開源RocketMQ支持延遲消息，但是不支持秒級精度。默認支持18個level的延遲消息，這是通過broker端的messageDelayLevel配置項確定的，如下：

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

Broker在啟動時，內部會創建一個內部主題：SCHEDULE_TOPIC_XXXX，根據延遲level的個數，創建對應數量的隊列，也就是說18個level對應了18個隊列。注意，這並不是說這個內部主題只會有18個隊列，因為Broker通常是集群模式部署的，因此每個節點都有18個隊列。

延遲級別的值可以進行修改，以滿足自己的業務需求，可以修改/添加新的level。例如：你想支持2天的延遲，修改最后一個level的值為2d，這個時候依然是18個level；也可以增加一個2d，這個時候總共就有19個level。

生產者發送延遲消息

生產者在發送延遲消息非常簡單，只需要設置一個延遲級別即可，注意不是具體的延遲時間，如：

Message msg=new Message();
msg.setTopic("TopicA");
msg.setTags("Tag");
msg.setBody("this is a delay message".getBytes());
//設置延遲level為5，對應延遲1分鍾
msg.setDelayTimeLevel(5);
producer.send(msg);

如果設置的延遲level超過最大值，那么將會重置最最大值。

Broker端存儲延遲消息

延遲消息在RocketMQ Broker端的流轉如下圖所示：

可以看到，總共有6個步驟，下面會對這6個步驟進行詳細的講解：

修改消息Topic名稱和隊列信息
轉發消息到延遲主題的CosumeQueue中
延遲服務消費SCHEDULE_TOPIC_XXXX消息
將信息重新存儲到CommitLog中
將消息投遞到目標Topic中
消費者消費目標topic中的數據

第一步：修改消息Topic名稱和隊列信息

RocketMQ Broker端在存儲生產者寫入的消息時，首先都會將其寫入到CommitLog中。之后根據消息中的Topic信息和隊列信息，將其轉發到目標Topic的指定隊列(ConsumeQueue)中。

由於消息一旦存儲到ConsumeQueue中，消費者就能消費到，而延遲消息不能被立即消費，所以這里將Topic的名稱修改為SCHEDULE_TOPIC_XXXX，並根據延遲級別確定要投遞到哪個隊列下。

同時，還會將消息原來要發送到的目標Topic和隊列信息存儲到消息的屬性中。相關源碼如下所示：
org.apache.rocketmq.store.CommitLog#asyncPutMessage

public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {

//如果是延遲消息
            if (msg.getDelayTimeLevel() > 0) {
                //如果設置的級別超過了最大級別 重置延遲級別
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }

                //修改TOPIC的投遞目標為內部主題SCHEDULE_TOPIC_XXX
                topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                //根據delayLevel 確定消息投遞到SCHEDULE_TOPIC_XXX內部的哪個隊列中
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                // Backup real topic, queueId
                //記錄原始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()));

                //更新消息的topic、queueId
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }

}

第二步：轉發消息到延遲主題的CosumeQueue中

CommitLog中的消息轉發到CosumeQueue中是異步進行的。在轉發過程中，會對延遲消息進行特殊處理，主要是計算這條延遲消息需要在什么時候進行投遞。

投遞時間=消息存儲時間(storeTimestamp) + 延遲級別對應的時間

需要注意的是，會將計算出的投遞時間當做消息Tag的哈希值存儲到CosumeQueue中，CosumeQueue單個存儲單元組成結構如下圖所示：

其中：

Commit Log Offset：記錄在CommitLog中的位置。
Size：記錄消息的大小
Message Tag HashCode：記錄消息Tag的哈希值，用於消息過濾。特別的，對於延遲消息，這個字段記錄的是消息的投遞時間戳。這也是為什么java中hashCode方法返回一個int型，只占用4個字節，而這里Message Tag HashCode字段卻設計成8個字節的原因。

相關源碼參見：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 (TopicValidator.RMQ_SYS_SCHEDULE_TOPIC.equals(topic) && t != null) {
                        int delayLevel = Integer.parseInt(t);

                        if (delayLevel > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                            delayLevel = this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel();
                        }

                        //如果延遲級別大於0 計算目標投遞時間 並將其當作tag哈希值
                        if (delayLevel > 0) {
                            tagsCode = this.defaultMessageStore.getScheduleMessageService().computeDeliverTimestamp(delayLevel,
                                storeTimestamp);
                        }
                    }
                }

}

第三步：延遲服務消費SCHEDULE_TOPIC_XXXX消息

Broker內部有一個ScheduleMessageService類，其充當延遲服務，消費SCHEDULE_TOPIC_XXXX中的消息，並投遞到目標Topic中。

ScheduleMessageService在啟動時，其會創建一個定時器Timer，並根據延遲級別的個數，啟動對應數量的TimerTask，每個TimerTask負責一個延遲級別的消費與投遞。

相關源碼如下所示：ScheduleMessageService#start

public void start() {
        if (started.compareAndSet(false, true)) {
            super.load();
            //1.創建定時器Timer
            this.timer = new Timer("ScheduleMessageTimerThread", true);
            //2.針對每個延遲級別 創建一個 TimerTask
            //2.1: 迭代每個延遲級別，delayLevelTable是一個Map 記錄了每個延遲級別對應的延遲時間
            for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
                Integer level = entry.getKey();
                Long timeDelay = entry.getValue();
                Long offset = this.offsetTable.get(level);
                if (null == offset) {
                    offset = 0L;
                }
                //2.2 針對每個延遲級別 創建一個 TimerTask
                if (timeDelay != null) {
                    this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
                }
            }
}

需要注意的是，每個TimeTask在檢查消息是否到期時，首先檢查對應隊列中尚未投遞第一條消息，如果這條消息沒到期，那么之后的消息都不會檢查。如果到期了，則進行投遞，並檢查之后的消息是否到期。

第四步：將信息重新存儲到CommitLog中

在將消息到期后，需要投遞到目標Topic。由於在第一步已經記錄了原來的Topic和隊列信息，因此這里重新設置，再存儲到CommitLog即可。此外，由於之前Message Tag HashCode字段存儲的是消息的投遞時間，這里需要重新計算tag的哈希值后再存儲。

源碼參見：DeliverDelayedMessageTimerTask的messageTimeup方法。

第五步：將消息投遞到目標Topic中

這一步與第二步類似，不過由於消息的Topic名稱已經改為了目標Topic。因此消息會直接投遞到目標Topic的ConsumeQueue中，之后消費者即消費到這條消息。

延遲消息與消費重試的關系

RocketMQ提供了消息重試的能力，在並發模式消費消費失敗的情況下，可以返回一個枚舉值RECONSUME_LATER，那么消息之后將會進行重試。如：

consumer.registerMessageListener(new MessageListenerConcurrently() {
       @Override
       public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                       ConsumeConcurrentlyContext context) {
           //處理消息，失敗，返回RECONSUME_LATER，進行重試
           return ConsumeConcurrentlyStatus.RECONSUME_LATER;
       }
   });

重試默認會進行重試16次。使用過RocketMQ消息重試功能的用戶，可能看到過以下這張圖：

第幾次重試	與上次重試的間隔時間	第幾次重試	與上次重試的間隔時間
1	10 秒	9	7 分鍾
2	30 秒	10	8 分鍾
3	1 分鍾	11	9 分鍾
4	2 分鍾	12	10 分鍾
5	3 分鍾	13	20 分鍾
6	4 分鍾	14	30 分鍾
7	5 分鍾	15	1 小時
8	6 分鍾	16	2 小時

細心地的讀者發現了，消息重試的16個級別，實際上是把延遲消息18個級別的前兩個level去掉了。事實上，RocketMQ的消息重試也是基於延遲消息來完成的。在消息消費失敗的情況下，將其重新當做延遲消息投遞回Broker。

在投遞回去時，會跳過前兩個level，因此只重試16次。當然，消息重試還有一些其他的設計邏輯，在之后的文章將會進行分析。

自定義延遲時間

開源版本延遲消息缺點:固定了Level，不夠靈活，最多只能支持18個Level

Java中的延遲任務

Timer

public static void main(String[] args) {
        Timer timer = new Timer();
        //在3秒后執行run方法，之后每隔1秒執行一次run方法
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("執行任務");
            }
        }, 3000);
    }

ScheduledThreadPoolExecutor

public static void main(String[] args) {
        ScheduledExecutorService scheduledExecutorService
                = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.schedule(
                () -> System.out.println("執行任務"), 3000,
                TimeUnit.MILLISECONDS);
    }

其原理都是基於最小堆實現的延遲隊列DelayQueue

插入任務的時間復雜度為Olog(n)，消息TPS較高時性能仍不夠快，有沒O(1)復雜度的方案呢？

TimeWheel時間輪

Netty、Kafka中使用TimeWheel來優化I/O超時的操作

ticksPerWheel：槽位數
tick：每個槽位的時間間隔

假設這個延遲時間為X秒，那么X%(ticksPerWheel * tick)可以計算出X所屬的TimeWheel中位置

TimeWheel的size為8，那么延遲1秒和9秒的消息都處在一個鏈表中。如果用戶先發了延遲9秒的消息再發了延遲1秒的消息，他們在一個鏈表中所以延遲1秒的消息會需要等待延遲9秒的消息先投遞。顯然這是不能接受的，那么如何解決這個問題？

排序
顯然，如果對TimeWheel一個tick中的元素進行排序顯然就解決了上面的問題。但是顯而易見的是排序是不可能的。
擴大時間輪
最直觀的方式，我們能不能通過擴大時間輪的方式避免延遲9和延遲1落到一個tick位置上？
假設支持30天，精度為1秒，那么ticksPerWheel=30 * 24 * 60 * 60，這樣每一個tick上的延遲都是一致的，不存在上述的問題（類似於將RocketMQ的Level提升到了30 * 24 * 60 * 60個）。但是TimeWheel需要被加載到內存操作，這顯然是無法接受的。
多級時間輪
單個TimeWheel無法支持，那么能否顯示中的時針、分針的形式，構建多級時間輪來解決呢？

多級時間輪解決了上述的問題，但是又引入了新的問題：

在整點（tick指向0的位置）需要加載大量的數據會導致延遲，比如第二個時間輪到整點需要加載未來一天的數據時間輪需要載入到內存，這個開銷是不可接受的

延遲加載
多級定時輪的問題在於需要加載大量數據到內存，那么能否優化一下將這里的數據延遲加載到內存來解決內存開銷的問題呢？
在多級定時輪的方案中，顯然對於未來一小時或者未來一天的數據可以不加載到內存，而可以只加載延遲時間臨近的消息。
進一步優化，可以將數據按照固定延遲間隔划分，那么每次加載的數據量是大致相同的，不會出tick約大的定時輪需要加載越多的數據，那么方案如下：

基於上述的方案，那么TimeWheel中存儲未來30分鍾需要投遞的消息的索引，索引為一個long型，那么數據量為：30 * 60 * 8 * TPS，相對來說內存開銷是可以接受的，比如TPS為1w那么大概開銷為200M+。
之后的數據按照每30分鍾一個塊的形式寫入文件，那么每個整點時的操作就是計算一下將30分鍾的消息Hash到對應的TimeWheel上，那么排序問題就解決了。
到此為止就只剩下一個問題，如何保存30天的數據？

CommitLog保存超長延遲的數據

CommitLog是有時效性的，比如在我們只保存最近7天的消息，過期數據將被刪除。對於延遲消息，可能需要30天之后投遞，顯然是不能被刪除的。

那么我們怎么保存延遲消息呢？

直觀的方法就是將延遲消息從CommitLog中剝離出來，獨立存儲以保存更長的時間。

通過DispatchService將WAL中的延遲消息寫入到獨立的文件中。這些文件按照延遲時間組成一個鏈表。

鏈表長度為最大延遲時間/每個文件保存的時間長度。
那么WAL可以按照正常的策略進行過期刪除，Delay Msg File則在一個文件投遞完之后進行刪除。

唯一的問題是這里會有Delay Msg File帶來的隨機寫問題，但是這個對系統整體性能不會有很大影響，在可接受范圍內。

BOUNS

結合TimeWheel和CommitLog保存超長延遲數據的方案，加上一些優化手段，基本就完成了支持任意延遲時間的

方案：

消息寫入WAL
Dispatcher處理延遲消息
延遲消息一定時間的直接寫入TimeWheel
延遲超過一定時間寫入DelayMessageStorage
DelayMessageStorage對DelayMsgFile構建一層索引，這樣在映射到TimeWheel時只需要做一次Hash操作
通過TimeWheel將消息投遞到ConsumeQueue中完成對Consumer的可見

本文轉載：
https://cloud.tencent.com/developer/article/1581368
https://www.cnblogs.com/hzmark/p/mq-delay-msg.html

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 深入理解RocketMQ（四）--消息存儲深入理解RocketMQ消息查詢機制深入理解RocketMQ（一）---閱讀源碼准備 RocketMQ定時（延遲）消息消息隊列模型深入理解深入理解 Handler 消息機制深入理解Mysql數據庫主從延遲深入理解JVM（③）低延遲的Shenandoah收集器深入理解RocketMQ的消費者組、隊列、Broker，Topic RocketMQ系列（五）廣播與延遲消息