文章目錄

• 1. 什么是延遲消息
• • 1.1 延時消息的使用場景
• 2. 示例
• 3. 原理
• 參考

1. 什么是延遲消息

發送消息后，消費者要等待一定的時間才能消費到該消息。

RocketMQ 不支持任意時間自定義的延遲消息，僅支持內置預設值的延遲時間間隔的延遲消息。

預設值的延遲時間間隔為：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h

對于開源版是這樣，而對于專業版，是可以自定義時間的。

1.1 延時消息的使用場景

比如電商里，提交了一個訂單就可以發送一個延時消息，1h后去檢查這個訂單的狀態，如果還是未付款就取消訂單釋放庫存。

2. 示例

對于生產者，設定一個延遲時間，而對于消費者，沒有什么不同：

生產者：

package com.xin.rocketmq.demo.testrun;import com.xin.rocketmq.demo.config.JmsConfig; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message;public class ProducerDelay {public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");producer.setNamesrvAddr("192.168.10.11:9876");producer.start();Message msg1 = new Message(JmsConfig.TOPIC,"訂單001".getBytes());msg1.setDelayTimeLevel(2);//延遲5秒Message msg2 = new Message(JmsConfig.TOPIC,"訂單001".getBytes());msg2.setDelayTimeLevel(4);//延遲30秒SendResult sendResult1 = producer.send(msg1);SendResult sendResult2 = producer.send(msg2);System.out.println("Product1-同步發送-Product信息={}" + sendResult1);System.out.println("Product2-同步發送-Product信息={}" + sendResult2);producer.shutdown();} }

msg2.setDelayTimeLevel(4);設置延遲時間。

消費者：

package com.xin.rocketmq.demo.testrun;import com.xin.rocketmq.demo.config.JmsConfig; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;public class ConsumerDelay {public static void main(String[] args) throws Exception {// 實例化消費者DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");// 設置NameServer的地址consumer.setNamesrvAddr("192.168.10.11:9876");// 訂閱一個或者多個Topic，以及Tag來過濾需要消費的消息consumer.subscribe(JmsConfig.TOPIC, "*");// 注冊消息監聽者consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {// Print approximate delay time periodSystem.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later");}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 啟動消費者consumer.start();} }

3. 原理

開源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。

可以看到這里并不支持秒級精度，按照《rocketmq developer guide》中的說法，是為了避免在broker對消息進行排序，造成性能影響。不過筆者考慮，之所以不支持，更多應該是商業上的考慮。

參考

深入理解RocketMQ延遲消息

總結

以上是生活随笔為你收集整理的【RocketMQ】延迟消息(延迟队列)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。