RocketMQ簡介
RocketMQ是一款開源的分布式消息系統，基於高可用分布式集群技術，提供低延時的、高可靠、萬億級容量、靈活可伸縮的消息發布與訂閱服務。
它前身是MetaQ，是阿里基於Kafka的設計使用Java進行自主研發的。在2012年，阿里將其開源， 在2016年，阿里將其捐獻給Apache軟件基金會（Apache Software Foundation，簡稱為ASF），正式成為孵化項目。2017 年，Apache軟件基金會宣布RocketMQ已孵化成為 Apache頂級項目（Top Level Project，簡稱為TLP ），是國內首個互聯網中間件在 Apache上的頂級項目。
延遲消息
生產者把消息發送到消息隊列中以后，並不期望被立即消費，而是等待指定時間后才可以被消費者消費，這類消息通常被稱為延遲消息。
在RocketMQ中，支持延遲消息，但是不支持任意時間精度的延遲消息，只支持特定級別的延遲消息。如果要支持任意時間精度，不能避免在Broker層面做消息排序，再涉及到持久化的考量，那么消息排序就不可避免產生巨大的性能開銷。
消息延遲級別分別為1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，共18個級別。在發送消息時，設置消息延遲級別即可，設置消息延遲級別時有以下3種情況：
- 設置消息延遲級別等於0時，則該消息為非延遲消息。
- 設置消息延遲級別大於等於1並且小於等於18時，消息延遲特定時間，如：設置消息延遲級別等於1，則延遲1s；設置消息延遲級別等於2，則延遲5s，以此類推。
- 設置消息延遲級別大於18時，則該消息延遲級別為18，如：設置消息延遲級別等於20，則延遲2h。
文章持續更新，微信搜索「萬貓學社第一時間閱讀，關注后回復「電子書」，免費獲取12本Java必讀技術書籍。
延遲消息示例
首先，寫一個消費者，用於消費延遲消息：
public class Consumer {
public static void main(String[] args) throws MQClientException {
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
// 實例化消費者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OneMoreGroup");
// 設置NameServer的地址
consumer.setNamesrvAddr("localhost:9876");
// 訂閱一個或者多個Topic，以及Tag來過濾需要消費的消息
consumer.subscribe("OneMoreTopic", "*");
// 注冊回調實現類來處理從broker拉取回來的消息
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
System.out.printf("%s %s Receive New Messages:%n"
, sdf.format(new Date())
, Thread.currentThread().getName());
for (MessageExt msg : msgs) {
System.out.printf("\tMsg Id: %s%n", msg.getMsgId());
System.out.printf("\tBody: %s%n", new String(msg.getBody()));
}
// 標記該消息已經被成功消費
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 啟動消費者實例
consumer.start();
System.out.println("Consumer Started.");
}
}
再寫一個延遲消息的生產者，用於發送延遲消息：
public class DelayProducer {
public static void main(String[] args) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
// 實例化消息生產者Producer
DefaultMQProducer producer = new DefaultMQProducer("OneMoreGroup");
// 設置NameServer的地址
producer.setNamesrvAddr("localhost:9876");
// 啟動Producer實例
producer.start();
Message msg = new Message("OneMoreTopic"
, "DelayMessage", "This is a delay message.".getBytes());
//"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
//設置消息延遲級別為3，也就是延遲10s。
msg.setDelayTimeLevel(3);
// 發送消息到一個Broker
SendResult sendResult = producer.send(msg);
// 通過sendResult返回消息是否成功送達
System.out.printf("%s Send Status: %s, Msg Id: %s %n"
, sdf.format(new Date())
, sendResult.getSendStatus()
, sendResult.getMsgId());
// 如果不再發送消息，關閉Producer實例。
producer.shutdown();
}
}
運行生產者以后，就會發送一條延遲消息：
10:37:14.992 Send Status: SEND_OK, Msg Id: C0A8006D5AB018B4AAC216E0DB690000
10秒鍾后，消費者收到的這條延遲消息：
10:37:25.026 ConsumeMessageThread_1 Receive New Messages:
Msg Id: C0A8006D5AB018B4AAC216E0DB690000
Body: This is a delay message.
文章持續更新，微信搜索「萬貓學社第一時間閱讀，關注后回復「電子書」，免費獲取12本Java必讀技術書籍。
延遲消息的原理分析
以下分析的RocketMQ源碼的版本號是4.7.1，版本不同源碼略有差別。
CommitLog
在org.apache.rocketmq.store.CommitLog中，針對延遲消息做了一些處理：
// 延遲級別大於0，就是延時消息
if (msg.getDelayTimeLevel() > 0) {
// 判斷當前延遲級別，如果大於最大延遲級別，
// 就設置當前延遲級別為最大延遲級別。
if (msg.getDelayTimeLevel() > this.defaultMessageStore
.getScheduleMessageService().getMaxDelayLevel()) {
msg.setDelayTimeLevel(this.defaultMessageStore
.getScheduleMessageService().getMaxDelayLevel());
}
// 獲取延遲消息的主題，
// 其中RMQ_SYS_SCHEDULE_TOPIC的值為SCHEDULE_TOPIC_XXXX
topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
// 根據延遲級別獲取延遲消息的隊列Id，
// 隊列Id其實就是延遲級別減1
queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
// 備份真正的主題和隊列Id
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()));
// 設置延時消息的主題和隊列Id
msg.setTopic(topic);
msg.setQueueId(queueId);
}
可以看到，每一個延遲消息的主題都被暫時更改為SCHEDULE_TOPIC_XXXX，並且根據延遲級別延遲消息變更了新的隊列Id。接下來，處理延遲消息的就是org.apache.rocketmq.store.schedule.ScheduleMessageService。
ScheduleMessageService
ScheduleMessageService是由org.apache.rocketmq.store.DefaultMessageStore進行初始化的，初始化包括構造對象和調用load方法。最后，再執行ScheduleMessageService的start方法：
public void start() {
// 使用AtomicBoolean確保start方法僅有效執行一次
if (started.compareAndSet(false, true)) {
this.timer = new Timer("ScheduleMessageTimerThread", true);
// 遍歷所有延遲級別
for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
// key為延遲級別
Integer level = entry.getKey();
// value為延遲級別對應的毫秒數
Long timeDelay = entry.getValue();
// 根據延遲級別獲得對應隊列的偏移量
Long offset = this.offsetTable.get(level);
// 如果偏移量為null，則設置為0
if (null == offset) {
offset = 0L;
}
if (timeDelay != null) {
// 為每個延遲級別創建定時任務，
// 第一次啟動任務延遲為FIRST_DELAY_TIME，也就是1秒
this.timer.schedule(
new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
}
}
// 延遲10秒后每隔flushDelayOffsetInterval執行一次任務，
// 其中，flushDelayOffsetInterval默認配置也為10秒
this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
// 持久化每個隊列消費的偏移量
if (started.get()) ScheduleMessageService.this.persist();
} catch (Throwable e) {
log.error("scheduleAtFixedRate flush exception", e);
}
}
}, 10000, this.defaultMessageStore
.getMessageStoreConfig().getFlushDelayOffsetInterval());
}
}
遍歷所有延遲級別，根據延遲級別獲得對應隊列的偏移量，如果偏移量不存在，則設置為0。然后為每個延遲級別創建定時任務，第一次啟動任務延遲為1秒，第二次及以后的啟動任務延遲才是延遲級別相應的延遲時間。
然后，又創建了一個定時任務，用於持久化每個隊列消費的偏移量。持久化的頻率由flushDelayOffsetInterval屬性進行配置，默認為10秒。
文章持續更新，微信搜索「萬貓學社第一時間閱讀，關注后回復「電子書」，免費獲取12本Java必讀技術書籍。
定時任務
ScheduleMessageService的start方法執行之后，每個延遲級別都創建自己的定時任務，這里的定時任務的具體實現就在DeliverDelayedMessageTimerTask類之中，它核心代碼是executeOnTimeup方法之中，我們來看一下主要部分：
// 根據主題和隊列Id獲取消息隊列
ConsumeQueue cq =
ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(
TopicValidator.RMQ_SYS_SCHEDULE_TOPIC
, delayLevel2QueueId(delayLevel));
如果沒有獲取到對應的消息隊列，則在DELAY_FOR_A_WHILE（默認為100）毫秒后再執行任務。如果獲取到了，就繼續執行下面操作：
// 根據消費偏移量從消息隊列中獲取所有有效消息
SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
如果沒有獲取到有效消息，則在DELAY_FOR_A_WHILE（默認為100）毫秒后再執行任務。如果獲取到了，就繼續執行下面操作：
// 遍歷所有消息
for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
// 獲取消息的物理偏移量
long offsetPy = bufferCQ.getByteBuffer().getLong();
// 獲取消息的物理長度
int sizePy = bufferCQ.getByteBuffer().getInt();
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;
// 省略部分代碼...
}
如果當前消息不到消費的時間，則在countdown毫秒后再執行任務。如果到消費的時間，就繼續執行下面操作：
// 根據消息的物理偏移量和大小獲取消息
MessageExt msgExt =
ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
offsetPy, sizePy);
如果獲取到消息，則繼續執行下面操作：
// 重新構建新的消息，包括：
// 1.清除消息的延遲級別
// 2.恢復真正的消息主題和隊列Id
MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) {
log.error("[BUG] the real topic of schedule msg is {},"
+ " discard the msg. msg={}",
msgInner.getTopic(), msgInner);
continue;
}
// 重新把消息發送到真正的消息隊列上
PutMessageResult putMessageResult =
ScheduleMessageService.this.writeMessageStore
.putMessage(msgInner);
清除了消息的延遲級別，並且恢復了真正的消息主題和隊列Id，重新把消息發送到真正的消息隊列上以后，消費者就可以立即消費了。
文章持續更新，微信搜索「萬貓學社第一時間閱讀，關注后回復「電子書」，免費獲取12本Java必讀技術書籍。
總結
經過以上對源碼的分析，可以總結出延遲消息的實現步驟：
- 如果消息的延遲級別大於0，則表示該消息為延遲消息，修改該消息的主題為SCHEDULE_TOPIC_XXXX，隊列Id為延遲級別減1。
- 消息進入SCHEDULE_TOPIC_XXXX的隊列中。
- 定時任務根據上次拉取的偏移量不斷從隊列中取出所有消息。
- 根據消息的物理偏移量和大小再次獲取消息。
- 根據消息屬性重新創建消息，清除延遲級別，恢復原主題和隊列Id。
- 重新發送消息到原主題的隊列中，供消費者進行消費。
微信公眾號：萬貓學社
微信掃描二維碼
關注后回復「 電子書」
獲取12本Java必讀技術書籍
