【RocketMQ】RocketMQ 5.0版本任意时刻延迟消息的实现原理浅析

太过爱你忘了你带给我的痛 2023-09-29 21:08 6阅读 0赞

#### 文章目录 ####

*  意外发现
 *  设计方案
 *   *  时间轮
     *  定时消息存储
 *  具体实现
 *   *  流程图
     *  流程步骤

## 意外发现 ##

无意间从官方的最新的客户端代码中看到下面的Example：

感兴趣的可以看看这个介绍：https://rocketmq.apache.org/docs/featureBehavior/02delaymessage

生产者：

// Send delay messages.
            MessageBuilder messageBuilder = null;
            // Specify a millisecond-level Unix timestamp. In this example, the specified timestamp indicates that the message will be delivered in 10 minutes from the current time. 
            Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
            Message message = messageBuilder.setTopic("topic")
                    // Specify the message index key. The system uses the key to locate the message. 
                    .setKeys("messageKey")
                    // Specify the message tag. The consumer can use the tag to filter messages. 
                    .setTag("messageTag")
                    .setDeliveryTimestamp(deliverTimeStamp)
                    // Configure the message body.
                    .setBody("messageBody".getBytes())
                    .build();
            try {
        
                // Send the messages. Focus on the result of message sending and exceptions such as failures. 
                SendReceipt sendReceipt = producer.send(message);
                System.out.println(sendReceipt.getMessageId());
            } catch (ClientException e) {
        
                e.printStackTrace();
            }

可以看到一句：setDeliveryTimestamp(deliverTimeStamp)，也就是说可以支持任意时刻的延迟消息了？？？

消费者：

// Consumption example 1: If a scheduled message is consumed by a push consumer, the consumer needs to process the message only in the message listener. 
            MessageListener messageListener = new MessageListener() {
        
                @Override
                public ConsumeResult consume(MessageView messageView) {
        
                    System.out.println(messageView.getDeliveryTimestamp());
                    // Return the status based on the consumption result. 
                    return ConsumeResult.SUCCESS;
                }
            };
            // Consumption example 2: If a scheduled message is consumed by a simple consumer, the consumer must obtain the message for consumption and submit the consumption result. 
            List<MessageView> messageViewList = null;
            try {
        
                messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
                messageViewList.forEach(messageView -> {
        
                    System.out.println(messageView);
                    // After consumption is complete, the consumer must invoke ACK to submit the consumption result. 
                    try {
        
                        simpleConsumer.ack(messageView);
                    } catch (ClientException e) {
        
                        e.printStackTrace();
                    }
                });
            } catch (ClientException e) {
        
                // If the pull fails due to system traffic throttling or other reasons, you must re-initiate the request to obtain the message. 
                e.printStackTrace();
            }

RocketMQ5.X版本新增了Proxy模块，从配置中可以看到，延迟消息方案**目前**默认还是通过按level来指定的，也就是说，可以选择不按level来执行了！！！

![f939c8ed0df71f71033a17b1c09d0ab7.png][]

下面一起看看新版本是怎么实现任意时刻延迟消息的。

## 设计方案 ##

### 时间轮 ###

首先，RocketMQ对任意时刻延迟消息的支持，是基于主流的方案——时间轮做的，时间轮，对时刻表的一种抽象，通常使用数组实现。时刻表上的每一秒，顺序对应到数组中的位置，然后数组循环使用。时间轮的每一格，指向了TimerLog中的对应位置，如果这一格的时间到了，则按TimerLog中的对应位置以及prev\_pos位置依次读出每条消息。时间轮一格一格向前推进，配合TimerLog，依次读出到期的消息，从而达到定时消息的目的。

时间轮的每一格设计如下：

![fc984eba785bcbc03cf0bf7d4f7ad524.png][]

### 定时消息存储 ###

定时消息的记录文件，Append Only。每条记录包含一个prev\_pos，指向前一条**定时到同样时刻**的记录。每条记录的内容可以包含定时消息本身，也可以只包含定时消息的位置信息。每一条记录包含如下信息：

![df2a55a0b40fc4d420410beb5f423b06.png][]![548f49a14d773a967f16e41b334fb9e3.png][]

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>大小</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>size</td> 
   <td>4B</td> 
   <td>保存记录的大小</td> 
  </tr> 
  <tr> 
   <td>prev_pos</td> 
   <td>8B</td> 
   <td>前一条记录的位置</td> 
  </tr> 
  <tr> 
   <td>current_time</td> 
   <td>8B</td> 
   <td>当前时间</td> 
  </tr> 
  <tr> 
   <td>magic</td> 
   <td>4B</td> 
   <td>magic value</td> 
  </tr> 
  <tr> 
   <td>delayed_time</td> 
   <td>4B</td> 
   <td>该条记录的定时时间</td> 
  </tr> 
  <tr> 
   <td>offset_real</td> 
   <td>8B</td> 
   <td>该条消息在commitLog中的位置</td> 
  </tr> 
  <tr> 
   <td>size_real</td> 
   <td>4B</td> 
   <td>该条消息在commitLog中的大小</td> 
  </tr> 
  <tr> 
   <td>hash_topic</td> 
   <td>4B</td> 
   <td>该条消息topic的hash code</td> 
  </tr> 
  <tr> 
   <td>varbody</td> 
   <td>8B</td> 
   <td>存储可变的body，暂时没有为空</td> 
  </tr> 
 </tbody> 
</table>

## 具体实现 ##

### 流程图 ###

![a6ad98c4808060b5861747418dbe8bf5.png][]

### 流程步骤 ###

从图中可以看出，共有五个Service分别处理定时消息的放置和存储。工作流如下：

1.  针对放置定时消息的service，每50ms从commitLog读取指定主题(rmq\_sys\_wheel\_timer)的定时消息。
    
    a. TimerEnqueueGetService从commitLog读取得到定时主题的消息，并先将其放入enqueuePutQueue。
    
    org.apache.rocketmq.store.timer.TimerMessageStore.TimerEnqueueGetService\#run
    
    ![b29094654e357f8b919033aff1aaa3ed.png][]
    
    org.apache.rocketmq.store.timer.TimerMessageStore\#enqueue
    
    ![6696b51bfca1f940f291052c542b223e.png][]
    
    b. 另一个线程TimerEnqueuePutService将其放入timerLog,更新时间轮的存储内容。将该任务放进时间轮的指定位置。
    
    org.apache.rocketmq.store.timer.TimerMessageStore.TimerEnqueuePutService\#run
    
    ![2d8cf3d4a7c9825ed21061829b03ddd5.png][]
    
    org.apache.rocketmq.store.timer.TimerMessageStore\#doEnqueue  
    ![e7484411c2035fbe1ad01583cf45fc67.png][]
2.  针对取出定时消息的service，每50ms读取下一秒的slot。有三个线程将读取到的消息重新放回commitLog。
    
    a. 首先，TimerDequeueGetService每50ms读一次下一秒的slot，从timerLog中得到指定的msgs，并放进dequeueGetQueue。
    
    org.apache.rocketmq.store.timer.TimerMessageStore.TimerDequeueGetService\#run
    
    ![af14f31c936a3fdf20ebf28a8b8ca685.png][]
    
    org.apache.rocketmq.store.timer.TimerMessageStore\#dequeue
    
    ![8c7ff57edd57ce089458802e74f42808.png][]  
    ![1211579dc7718ca1c6b3a6f7aa185345.png][]
    
    b. 而后TimerDequeueGetMessageService从dequeueGetQueue中取出msg，并将其放入队列中。该队列为待写入commitLog的队列，dequeuePutQueue。
    
    org.apache.rocketmq.store.timer.TimerMessageStore.TimerDequeueGetMessageService\#run
    
    ![461fbd95cd4531a38ff46efff847f2cc.png][]
    
    c. 最后TimerDequeuePutMessageService将这个queue中的消息取出，若已到期则修改topic，放回commitlog，否则继续按原topic写回CommitLog滚动。
    
    org.apache.rocketmq.store.timer.TimerMessageStore.TimerDequeuePutMessageService\#run
    
    ![6ac910d4b78cd118e91dd9bcc31ffd92.png][]
    
    org.apache.rocketmq.store.timer.TimerMessageStore\#convertMessage
    
    消息转换，更改真实Topic  
    ![fb4091c6ad95de107fea65083a4c7113.png][]
    
    投递消息
    
    org.apache.rocketmq.store.timer.TimerMessageStore\#doPut
    
    ![46c0480a9e0890dc5ff83501550a5760.png][]

消息投递到真实Topic后，其实就变成了一条“正常的消息”了，消费者就能正常消费了，以上就是对RocketMQ 5.0中延迟消息的变更做的分析，参考了部分官方的资料，后续会使用5.0版本，实际做一些演练，目前对于这个新特性，官方并没有大肆宣扬，也不知道具体有哪些限制，所以还需要做一些实践，踩踩坑。

[f939c8ed0df71f71033a17b1c09d0ab7.png]: https://img-blog.csdnimg.cn/img_convert/f939c8ed0df71f71033a17b1c09d0ab7.png
[fc984eba785bcbc03cf0bf7d4f7ad524.png]: https://img-blog.csdnimg.cn/img_convert/fc984eba785bcbc03cf0bf7d4f7ad524.png
[df2a55a0b40fc4d420410beb5f423b06.png]: https://img-blog.csdnimg.cn/img_convert/df2a55a0b40fc4d420410beb5f423b06.png
[548f49a14d773a967f16e41b334fb9e3.png]: https://img-blog.csdnimg.cn/img_convert/548f49a14d773a967f16e41b334fb9e3.png
[a6ad98c4808060b5861747418dbe8bf5.png]: https://img-blog.csdnimg.cn/img_convert/a6ad98c4808060b5861747418dbe8bf5.png
[b29094654e357f8b919033aff1aaa3ed.png]: https://img-blog.csdnimg.cn/img_convert/b29094654e357f8b919033aff1aaa3ed.png
[6696b51bfca1f940f291052c542b223e.png]: https://img-blog.csdnimg.cn/img_convert/6696b51bfca1f940f291052c542b223e.png
[2d8cf3d4a7c9825ed21061829b03ddd5.png]: https://img-blog.csdnimg.cn/img_convert/2d8cf3d4a7c9825ed21061829b03ddd5.png
[e7484411c2035fbe1ad01583cf45fc67.png]: https://img-blog.csdnimg.cn/img_convert/e7484411c2035fbe1ad01583cf45fc67.png
[af14f31c936a3fdf20ebf28a8b8ca685.png]: https://img-blog.csdnimg.cn/img_convert/af14f31c936a3fdf20ebf28a8b8ca685.png
[8c7ff57edd57ce089458802e74f42808.png]: https://img-blog.csdnimg.cn/img_convert/8c7ff57edd57ce089458802e74f42808.png
[1211579dc7718ca1c6b3a6f7aa185345.png]: https://img-blog.csdnimg.cn/img_convert/1211579dc7718ca1c6b3a6f7aa185345.png
[461fbd95cd4531a38ff46efff847f2cc.png]: https://img-blog.csdnimg.cn/img_convert/461fbd95cd4531a38ff46efff847f2cc.png
[6ac910d4b78cd118e91dd9bcc31ffd92.png]: https://img-blog.csdnimg.cn/img_convert/6ac910d4b78cd118e91dd9bcc31ffd92.png
[fb4091c6ad95de107fea65083a4c7113.png]: https://img-blog.csdnimg.cn/img_convert/fb4091c6ad95de107fea65083a4c7113.png
[46c0480a9e0890dc5ff83501550a5760.png]: https://img-blog.csdnimg.cn/img_convert/46c0480a9e0890dc5ff83501550a5760.png