RocketMQ 延遲消息
RocketMQ 消費者啟動流程
什么是延遲消息
RocketMQ 延遲消息是指，生產者發送消息給消費者消息，消費者需要等待一段時間后才能消費到。
使用場景
用戶下單之后，15分鐘未支付，對支付賬單進行提醒或者關單處理。
RocketMQ 開源版本的消息不支持任意時間精度，只支持5s 10s 1m等等。
Broker 如何處理延遲消息
消息投遞如下：
- 生產者發送一個延遲消息到一個
topic中 - Broker 判斷是個延遲消息后，將消息暫存
- Broker 通過延遲服務， 先檢查消息是否過期，如果到期將消息投遞到目標
topic - 消費者消費
topic中的投遞延遲消息。
開源RocketMQ 的消息不支持任意精度，默認支持 18個 level：
messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
Broker 在啟動的時候，會創建一個內部 topic:“SCHEDULE_TOPIC_XXXX” 根據延遲 level 數量，創建對應數量的 隊列。 也就是說 18 level 對應了18 個隊列。
具體可以在 代碼TopicConfigManager.java 中 看到：
private static final int SCHEDULE_TOPIC_QUEUE_NUM = 18;
要注意的是，Broker 一般是集群模式
部署，也就是說，每個Broker 都會有18個隊列。
TopicConfigManager#TopicConfigManager(BrokerController brokerController)
生產者消息延遲發送
代碼示例如下：
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);
Broker 存儲延遲消息
上一篇文章已經談到，Broker 收到消費者消息后，會進行消息存儲，然后再轉發到消費隊列（ConsumerQueue)，然后再推給消費者。
其實一旦消息轉發到
存儲延遲消息的流程也類似
- 確定延遲消息投遞到topic 哪個隊列。存儲生產者寫入的消息時，將消息轉發到 ConsumeQueue 中，消費者就能消費到。 延遲消息不能立即消息到，于是將 topic 名稱修改為
SCHEDULE_TOPIC_XXX，并根據延遲消息級別，確定投遞到哪個隊列上。同時還會將原來消息要發送到的目標 topic 和隊列記錄投遞到哪個隊列。
代碼在CommitLog#asyncPutMessage 中
設置延遲消息的投遞隊列信息代碼如下：
// Delay Deliveryif (msg.getDelayTimeLevel() > 0) {// 如果設置的級別超過了最大級別，重置延遲級別if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());}
// 計算延遲消息應該投遞到 SCHEDULE_TOPIC_XXXX 到哪個隊列。topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());// Backup real topic, queueId// 記錄原始 topic ,queueid,方便后期投遞到目標 topicMessageAccessor.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()));
// 更新消息投遞目標為 SCHEDULE_TOPIC_XXX,queueIdmsg.setTopic(topic);msg.setQueueId(queueId);}
消息轉發
消息轉發過程其實中會對延遲消息做一些特殊處理
CommitLog中的消息轉發到CosumeQueue中是異步進行的。在轉發過程中，會對延遲消息進行特殊處理，主要是計算這條延遲消息需要在什么時候進行投遞。
