2. 程式人生 > >RocketMQ學習(七)——RocketMQ延時訊息

RocketMQ學習(七)——RocketMQ延時訊息

阿新 • • 發佈：2019-02-08

使用者傳送的訊息也可以指定延時時間（更準確的說是延時等級），然後在指定延時時間之後投遞訊息，然後被consumer消費。阿里雲的ons還支援定時訊息，而且延時訊息是直接指定延時時間，其實阿里雲的延時訊息也是定時訊息的另一種表述方式，都是通過設定訊息被投遞的時間來實現的，但是Apache RocketMQ在版本4.2.0中尚不支援指定時間的延時，只能通過配置延時等級和延時等級對應的時間來實現延時。

一個延時訊息被髮出到消費成功經歷以下幾個過程：
1) 設定訊息的延時級別delayLevel。
2) producer傳送訊息。
3) broker收到訊息在準備將訊息寫入儲存的時候，判斷是延時訊息則更改Message的topic為延時訊息佇列的topic，也就是將訊息投遞到延時訊息佇列。
4)

有定時任務從延時佇列中讀取訊息，拿到訊息後判斷是否達到延時時間，如果到了則修改topic為原始topic。並將訊息投遞到原始topic的佇列。
5) consumer像消費其他訊息一樣從broker拉取訊息進行消費。
注意： 批量訊息是不支援延時訊息的。

producer傳送延時訊息

在producer中傳送訊息的時候，設定Message的delayLevel。

// org.apache.rocketmq.common.message.Message
public void setDelayTimeLevel(int level) {
    this.putProperty(MessageConst.
 
PROPERTY_DELAY_TIME_LEVEL, String.valueOf(level));
}

呼叫上面的方法設定延時等級的時候，會向message新增"DELAY"屬性，後面broker處理延時訊息就是依賴該屬性進行特別的處理。接下來發送訊息的流程和正常傳送訊息的流程基本一致，只是會將該訊息標記為延時訊息型別。

// org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendKernelImpl 
if (msg.getProperty("__STARTDELIVERTIME") != null || msg.
 
getProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL) != null) {
    context.setMsgType(MessageType.Delay_Msg);
}

broker處理延時訊息

broker收到延時訊息和正常訊息在前置的處理流程是一致的，對於延時訊息的特殊處理體現在將訊息寫入儲存（記憶體或檔案）的時候。

// org.apache.rocketmq.store.CommitLog#putMessage
public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
    // 省略中間程式碼...
    StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

    // 拿到原始topic和對應的queueId
    String topic = msg.getTopic();
    int queueId = msg.getQueueId();

    final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    // 非事務訊息和事務的commit訊息才會進一步判斷delayLevel
    if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
        || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        // Delay Delivery
        if (msg.getDelayTimeLevel() > 0) {
            // 糾正設定過大的level，就是delayLevel設定都大於延時時間等級的最大級
            if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
            }

            // 設定為延時佇列的topic
            topic = ScheduleMessageService.SCHEDULE_TOPIC;
            // 每一個延時等級一個queue，queueId = delayLevel - 1
            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()));
            // 更新properties
            msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

            msg.setTopic(topic);
            msg.setQueueId(queueId);
        }
    }
    // 省略中間程式碼...
}

上面的SCHEDULE_TOPIC是：

public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";

這個topic是一個特殊的topic，和正常的topic不同的地方是：
1、不會建立TopicConfig，因為也不需要consumer直接消費這個topic下的訊息。
2、不會將topic註冊到namesrv。
3、這個topic的佇列個數和延時等級的個數是相同的。
後面訊息寫入的過程和普通的又是一致的。上面將訊息寫入延時佇列中了，接下來就是處理延時佇列中的訊息，然後重新發送回原始topic的佇列中。
在此之前先說明下至今還有疑問的一個個概念——delayLevel。這個概念和我們接下要需要用到的的類ScheduleMessageService有關，這個類的欄位delayLevelTable裡面儲存了具體的延時等級。

private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable = new ConcurrentHashMap<Integer, Long>(32);

看下這個欄位的初始化過程：

// org.apache.rocketmq.store.schedule.ScheduleMessageService#parseDelayLevel
public boolean parseDelayLevel() {
    HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
    // 每個延時等級延時時間的單位對應的ms數
    timeUnitTable.put("s", 1000L);
    timeUnitTable.put("m", 1000L * 60);
    timeUnitTable.put("h", 1000L * 60 * 60);
    timeUnitTable.put("d", 1000L * 60 * 60 * 24);

    // 延時等級在MessageStoreConfig中配置
    // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
    String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
    try {
        // 根據空格將配置分隔出每個等級
        String[] levelArray = levelString.split(" ");
        for (int i = 0; i < levelArray.length; i++) {
            String value = levelArray[i];
            String ch = value.substring(value.length() - 1);
            // 時間單位對應的ms數
            Long tu = timeUnitTable.get(ch);

            // 延時等級從1開始
            int level = i + 1;
            if (level > this.maxDelayLevel) {
                // 找出最大的延時等級
                this.maxDelayLevel = level;
            }
            long num = Long.parseLong(value.substring(0, value.length() - 1));
            long delayTimeMillis = tu * num;
            this.delayLevelTable.put(level, delayTimeMillis);
    // 省略部分程式碼...
}

上面這個load方法在broker啟動的時候DefaultMessageStore會呼叫來初始化延時等級。接下來就應該解決怎麼處理延時訊息佇列中的訊息的問題了。處理延時訊息的服務是：ScheduleMessageService。是broker啟動的時候DefaultMessageStore會呼叫ScheduleMessageService#start來啟動處理延時訊息佇列的服務：

public void start() {
    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;
        }

        if (timeDelay != null) {
            // 每個延時佇列啟動一個定時任務來處理該佇列的延時訊息
            this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
        }
    }

    this.timer.scheduleAtFixedRate(new TimerTask() {

        @Override
        public void run() {
            try {
                // 持久化offsetTable(儲存了每個延時佇列對應的處理進度offset)
                ScheduleMessageService.this.persist();
            } catch (Throwable e) {
                log.error("scheduleAtFixedRate flush exception", e);
            }
        }
    }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
}

DeliverDelayedMessageTimerTask是一個TimerTask，啟動以後不斷處理延時佇列中的訊息，直到出現異常則終止該執行緒重新啟動一個新的TimerTask，它會呼叫executeOnTimeup方法。

public void executeOnTimeup() {
    // 找到該延時等級對應的ConsumeQueue
    ConsumeQueue cq =
        ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,
            delayLevel2QueueId(delayLevel));
    // 記錄異常情況下一次啟動TimerTask開始處理的offset
    long failScheduleOffset = offset;

    if (cq != null) {
        // 找到offset所處的MappedFile中offset後面的buffer
        SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
        if (bufferCQ != null) {
            try {
                long nextOffset = offset;
                int i = 0;
                ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
                for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                    // 下面三個欄位資訊是ConsumeQueue物理儲存的資訊
                    long offsetPy = bufferCQ.getByteBuffer().getLong();
                    int sizePy = bufferCQ.getByteBuffer().getInt();
                    // 注意這個tagCode，不再是普通的tag的hashCode，而是該延時訊息到期的時間
                    long tagsCode = bufferCQ.getByteBuffer().getLong();
                    // 省略中間程式碼....
                    long now = System.currentTimeMillis();
                    // 計算應該投遞該訊息的時間，如果已經超時則立即投遞
                    long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
                    // 計算下一個訊息的開始位置，用來尋找下一個訊息位置(如果有的話)
                    nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
                    // 判斷延時訊息是否到期
                    long countdown = deliverTimestamp - now;

                    if (countdown <= 0) {
                        MessageExt msgExt =
                            ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
                                offsetPy, sizePy);

                        if (msgExt != null) {
                            try {
                                // 將訊息恢復到原始訊息的格式，恢復topic、queueId、tagCode等，清除屬性"DELAY"
                                MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
                                PutMessageResult putMessageResult =
                                    ScheduleMessageService.this.defaultMessageStore
                                        .putMessage(msgInner);

                                if (putMessageResult != null
                                    && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
                                    // 投遞成功，處理下一個
                                    continue;
                                } else {
                                    // XXX: warn and notify me
                                    log.error(
                                        "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
                                        msgExt.getTopic(), msgExt.getMsgId());
                                    // 投遞失敗，結束當前task，重新啟動TimerTask，從下一個訊息開始處理，也就是說當前訊息丟棄
                                    // 更新offsetTable中當前佇列的offset為下一個訊息的offset
                                    ScheduleMessageService.this.timer.schedule(
                                        new DeliverDelayedMessageTimerTask(this.delayLevel,
                                            nextOffset), DELAY_FOR_A_PERIOD);
                                    ScheduleMessageService.this.updateOffset(this.delayLevel,
                                        nextOffset);
                                    return;
                                }
                            } catch (Exception e) {
                                // 重新投遞期間出現任何異常，結束當前task，重新啟動TimerTask，從當前訊息開始重試
                                /*
                                 * XXX: warn and notify me
                                 */
                                log.error(
                                    "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
                                        + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
                                        + offsetPy + ",sizePy=" + sizePy, e);
                            }
                        }
                    } else {
                        ScheduleMessageService.this.timer.schedule(
                            new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
                            countdown);
                        ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                        return;
                    }
                } // end of for
                // 處理完當前MappedFile中的訊息後，重新啟動TimerTask，從下一個訊息開始處理
                // 更新offsetTable中當前佇列的offset為下一個訊息的offset
                nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
                ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
                    this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
                ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                return;
            } finally {

                bufferCQ.release();
            }
        } // end of if (bufferCQ != null)
        else {
            // 如果根據offsetTable中的offset沒有找到對應的訊息(可能被刪除了)，則按照當前ConsumeQueue的最小offset開始處理
            long cqMinOffset = cq.getMinOffsetInQueue();
            if (offset < cqMinOffset) {
                failScheduleOffset = cqMinOffset;
                log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
                    + cqMinOffset + ", queueId=" + cq.getQueueId());
            }
        }
    } // end of if (cq != null)
    
    ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
        failScheduleOffset), DELAY_FOR_A_WHILE);
}

對於上面的tagCode做一下特別說明，延時訊息的tagCode和普通訊息不一樣：

  • 延時訊息的tagCode：儲存的是訊息到期的時間。
  • 非延時訊息的tagCode：tags字串的hashCode。

總結

以上就是RocketMQ延時訊息的實現方式，上面沒有詳說的是重試訊息的延時是怎麼實現的，其實就是在consumer將延時訊息傳送回broker的時候設定了（使用者可以自己設定，如果沒有自己設定預設是0）delayLevel，到了broker處理重試訊息的時候如果delayLevel是0（也就是說是預設的延時等級）的時候會在原來的基礎上加3，後面的處理就和上面說的延時訊息一樣了，儲存的時候將訊息投遞到延時佇列，等待延時到期後再重新投遞到原始topic佇列中等到consumer消費。

相關推薦

RocketMQ學習(七)——RocketMQ延時訊息

使用者傳送的訊息也可以指定延時時間（更準確的說是延時等級），然後在指定延時時間之後投遞訊息，然後被consumer消費。阿里雲的ons還支援定時訊息，而且延時訊息是直接指定延時時間，其實阿里雲的延時訊息也是定時訊息的另一種表述方式，都是通過設定訊息被投遞的時間

RocketMQ原理學習--延時訊息實現原理

         RocketMQ提供了延時訊息型別，簡單來說就是生產者在傳送訊息的時候指定一個延時時間，當到達延時時間之後訊息才能夠被投送到消費者。       首先我們可以考慮一下RocketMQ的延時訊息是

RocketMQ延時訊息的使用和延時級別的配置

RocketMQ 支援定時訊息，但是不支援任意時間精度，僅支援特定的 level，例如定時 5s， 10s， 1m 等。 其中，level=0 級表示不延時，level=1 表示 1 級延時，level=2 表示 2 級延時，以此類推。 如果要支援任意的時間精度，在Broker層面，必須要做訊息排序，

RocketMQ延時訊息

一個延時訊息被髮出到消費成功經歷以下幾個過程：設定訊息的延時級別delayLevelproducer傳送訊息broker收到訊息在準備將訊息寫入儲存的時候，判斷是延時訊息則更改Message的topic為延時訊息佇列的topic，也就是將訊息投遞到延時訊息佇列有定時任務從延時

Spring boot實戰專案整合阿里雲RocketMQ （非開源版）訊息佇列實現傳送普通訊息，延時訊息 --附程式碼

一.為什麼選擇RocketMQ訊息佇列？ 首先RocketMQ是阿里巴巴自研出來的，也已開源。其效能和穩定性從雙11就能看出來，借用阿里的一句官方介紹：歷年雙 11 購物狂歡節零點千萬級 TPS、萬億級資料洪峰，創造了全球最大的業務訊息併發以及流轉紀錄（日誌類訊息除外）；  在始終保證高效能前提下

運維學習之系統延時任務、定時任務以及臨時文件的管理

linux系統的延時及定時任務1.延時任務at 命令 發起的任務都是一次性的at +time下圖表示在21：22分進行刪除命令命令ctrl+d 表示發起動作at -l | atq #查看當前任務at -d | atrm #取消指定任務at -c #查看任務內容由圖二知主要執行touch這條命令at n

延時訊息佇列

下面程式碼按需要填寫 @Bean public Queue delayQueuePerMessageTTL() { Map<String, Object> argument = new HashMap<>(); argument.put(“x-message-ttl

RabbitMQ 延時訊息設計

 問題背景 所謂"延時訊息"是指當訊息被髮送以後，並不想讓消費者立即拿到訊息，而是等待指定時間後，消費者才拿到這個訊息進行消費。 場景一：客戶A在十二點下了一個訂單，我想半個小時後來檢查一下這個訂單的付款狀態，根據付款狀態來作下一步的處理。 a. 針對場景一，建議採用方案資料庫儲存+sc

Springboot2（28）整合rabbitmq實現延時訊息

原始碼地址 springboot2教程系列 rabbitmq實現訊息的確認機制和延時訊息的傳送 訊息生產者程式碼實現的主要配置 @Configuration @Slf4j public class PrividerRabbitm

Redis延時訊息佇列、非同步訊息佇列的實現

package list; import java.lang.reflect.Type; import java.util.Set; import java.util.UUID; import com.alibaba.fastjson.JSON; import com.a

rabbitmq發延時訊息以及通過一個exchange發到不同的queue

public static void main(String[] args) throws Exception { producer(1); producer(2); producer(3); } private static void

阿里MQ普通+順序+延時訊息 整合Spring

前言 由於公司專案需要，研究了下AliWareMQ。阿里mq的普通訊息和延時訊息還是挺簡單的。不過在順序訊息的時候出現了一些瓶頸。後來查閱原始碼和依據demo整理了一版融合Spring的版本。 例項 mq配置檔案(Spring) 主要是順序訊息的

Java關於DelayQueue做延時訊息推送

最近比較閒，看某專案原始碼時看到有用DelayQueue類來做延時的訊息推送。 DelayQueue是Delayed元素的一個無界阻塞佇列，只有在延遲期滿時才能從中提取元素。該佇列的頭部 是延遲期滿後儲存時間最長的Delayed元素。如果延遲都還沒有期滿，則佇列沒有頭部，並

Redis學習筆記之延時隊列

println itl bre type 列表 ring hist mat lib 目錄 一、業務場景 二、Redis延時隊列 一、業務場景 所謂延時隊列就是延時的消息隊列，

rocketmq學習(一) rocketmq介紹與安裝

1.訊息佇列介紹 　　訊息佇列本質上來說是一個符合先進先出原則的單向佇列：一方傳送訊息並存入訊息佇列尾部(生產者投遞訊息)，一方從訊息佇列的頭部取出訊息(消費者消費訊息)。但對於一個成熟可靠的訊息佇列來說，所需要解決的主要問題還包括：高效可靠的訊息投遞、儲存；能承受高併發的流量衝擊，可通過叢集部署來解決單點故

rocketmq學習(二) rocketmq叢集部署與圖形化控制檯安裝

1.rocketmq圖形化控制檯安裝 　　雖然rocketmq為使用者提供了使用命令列管理主題、消費組以及broker配置的功能，但對於不夠熟練的非運維人員來說，命令列的管理介面還是較難使用的。為此，我們可以使用圖形化的管理介面來簡化管理操作。 　　rocketmq官方推薦的圖形化控制檯目前還處在不成熟的孵化

手把手實現一條延時訊息

前言 近期在維護公司的排程平臺，其中有個關鍵功能那就是定時任務；定時任務大家平時肯定接觸的不少，比如 JDK 中的 Timer、ScheduledExecutorService、排程框架 Quartz 等。 通常用於實現 XX 時間後的延時任務，或週期性任務； 比如一個常見的業務場景：使用者下單 N 分鐘

玩轉redis-延時訊息佇列

上一篇基於redis的list實現了一個簡單的訊息佇列：玩轉redis-簡單訊息佇列 原始碼地址 使用demo 產品經理經常說的一句話，我們不光要有X功能，還要Y功能，這樣客戶才能更滿意。同樣的，只有簡單訊息佇列是不夠的，還要有延時訊息佇列才能算是一個完整的訊息佇列。 看看redis的命令，放眼望去，的有序

rocketmq延時消息

string 解決 print log void ride reg .sh art rocketmq提供一種延時消息的解決方案，就是在特定的時間到了，消息才會被投遞出去供consumer消費。總體來是簡單的場景是滿足了，但是需要註意的是延時的時間是需要按照默認配置的延時級別

訊息中介軟體學習總結（1）——RocketMQ之專訪RocketMQ聯合創始人：專案思路、技術細節和未來規劃

編者按 這些年開源氛圍越來越好，各大IT公司都紛紛將一些自研程式碼開源出來。2012年，阿里巴巴開源其自研的第三代分散式訊息中介軟體——RocketMQ。經過幾年的技術打磨，阿里稱基於RocketMQ技術，目前雙十一當天訊息容量可達到萬億級。 2016年11月，阿里將Ro