009rocketmq延时消息

文章目录 延迟消息第⼀步：修改消息Topic名称和队列信息第⼆步：转发消息到延迟主题的CosumeQueue中第三步：延迟服务消费SCHEDULE_TOPIC_XXXX消息第四步：将信息重新存储到CommitLog中第五步：将消息投递到目标Topic中第六步：消费者消费目标topic中的数据ScheduledMessageProducer.javaScheduledMessageConsumer.java

延迟消息

两个关键要素：临时存储 延时服务 内置队列，RocketMQ包含多个内置队列。 特点：对消费者不可见，消费者不能消费 SCHEDULE_TOPIC_XXXX对应的队列是一种内置队列

注意：RocketMQ不⽀持任意时间的延时，只⽀持以下⼏个固定的延时等级

String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

RocketMQ的延迟等级可以进⾏修改，以满⾜⾃⼰的业务需求，可以修改/添加新的level。 例如：你想⽀持1天的延迟，修改最后⼀个level的值为1d，这个时候依然是18个level；也可以增加⼀个1d，这个时候总共就有19个level。

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d

实现原理 延迟队列的核⼼思路是：所有的延迟消息由producer发出之后，都会存放到同⼀个topic（SCHEDULE_TOPIC_XXXX）下，根据延迟level的个数，创建对应数量的队列，也就是说18个level对应了18个队列。注意，这并不是说这个内部主题只会有18个队列，因为Broker通常是集群模式部署的，因此每个节点都有18个队列。不同的延迟级别会对应不同的队列序号，当延迟时间到之后，由 定时线程读取转换为普通的消息存的真实指定的topic下，此时对于consumer端此消息才可⻅，从⽽被consumer消费。 ⽣产者在发送延迟消息⾮常简单，只需要设置⼀个延迟级别即可，注意不是具体的延迟时间，如：

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);

延迟消息在RocketMQ Broker端的流转如下图所示：

可以看到，总共有6个步骤，下⾯会对这6个步骤进⾏详细的讲解：

修改消息Topic名称和队列信息转发消息到延迟主题SCHEDULE_TOPIC_XXXX的CosumeQueue中延迟服务消费SCHEDULE_TOPIC_XXXX消息将信息重新存储到CommitLog中将消息投递到⽬标Topic中消费者消费⽬标topic中的数据 第⼀步：修改消息Topic名称和队列信息

RocketMQ Broker端在存储⽣产者写⼊的消息时，⾸先都会将其写⼊到CommitLog中。之后根据消息中的Topic信息和队列信息，将其转发到⽬标Topic的指定队列(ConsumeQueue)中。 由于消息⼀旦存储到ConsumeQueue中，消费者就能消费到，⽽延迟消息不能被⽴即消费，所以这⾥将Topic的名称修改为SCHEDULE_TOPIC_XXXX，并根据延迟级别确定要投递到哪个队列下。 同时，还会将消息原来要发送到的⽬标Topic和队列信息存储到消息的属性中。

如果是延迟消息，修改topic，通过timelevel判断是延时消息 如果设置的级别超过了最大级别，重置延迟级别 更改topic 相同的延迟等级发送到同一个队列，不同等级不同队列 记录真正的topic和队列

/store/consumequeue/SCHEDULE_TOPIC_XXXX/2 相关源码如下所示： org.apache.rocketmq.store.CommitLog#putMessage

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {} 第⼆步：转发消息到延迟主题的CosumeQueue中

CommitLog中的消息转发到CosumeQueue中是异步进⾏的。在转发过程中，会对延迟消息进⾏特殊处理，主要是计算这条延迟消息需要在什么时候进⾏投递。

投递时间=消息存储时间(storeTimestamp) + 延迟级别对应的时间

需要注意的是，会将计算出的投递时间当做消息Tag的哈希值存储到CosumeQueue中，CosumeQueue单个存储单元组成结构如下图所示： 其中： Commit Log Offset：记录在CommitLog中的位置。 Size：记录消息的⼤⼩ Message Tag HashCode：记录消息Tag的哈希值，⽤于消息过滤。特别的，对于延迟消息，这个字段记录的是消息的投递时间戳。这也是为什么java中hashCode⽅法返回⼀个int型，只占⽤4个字节，⽽这⾥Message Tag HashCode字段却设计成8个字节的原因。 相关源码参⻅： CommitLog#checkMessageAndReturnSize

public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC, final boolean readBody) {} 第三步：延迟服务消费SCHEDULE_TOPIC_XXXX消息

Broker内部有⼀个ScheduleMessageService类，其充当延迟服务，消费SCHEDULE_TOPIC_XXXX中的消息，并投递到⽬标Topic中。 ScheduleMessageService在启动时，其会创建⼀个定时器Timer，并根据延迟级别的个数，启动对应数量的TimerTask，每个TimerTask负责⼀个延迟级别的消费与投递。 相关源码如下所示： ScheduleMessageService#start

public void start() {}

需要注意的是，每个TimeTask在检查消息是否到期时，⾸先检查对应队列中尚未投递第⼀条消息，如果这条消息没到期，那么之后的消息都不会检查。如果到期了，则进⾏投递，并检查之后的消息是否到期。

第四步：将信息重新存储到CommitLog中

在将消息到期后，需要投递到⽬标Topic。由于在第⼀步已经记录了原来的Topic和队列信息，因此这⾥重新设置，再存储到CommitLog即可。此外，由于之前Message Tag HashCode字段存储的是消息的投递时间，这⾥需要重新计算tag的哈希值后再存储。 源码参见：DeliverDelayedMessageTimerTask#messageTimeup⽅法

private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {} 第五步：将消息投递到目标Topic中 第六步：消费者消费目标topic中的数据

不同的延迟级别放在不同的队列序号下（queueId=delayLevel-1）。每⼀个延迟级别对应的延迟消息转换为普通消息的位置标识存放在~/store/config/delayOffset.json⽂件内。 key为对应的延迟级别，value对应不同延迟级别转换为普通消息的offset值。

{ "offsetTable":{3:202,4:2,5:2,6:2,7:2,8:2,9:2,10:2,11:2 } ScheduledMessageProducer.java package com.example.rocketmq.demo.scheduled; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq mon.message.Message; public class ScheduledMessageProducer { public static void main(String[] args) throws Exception { // Instantiate a producer to send scheduled messages DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup"); producer.setNamesrvAddr("localhost:9876"); // Launch producer producer.start(); for (int i = 0; i < 2; i++) { Message message = new Message("TopicTest", ("Hello scheduled message " + i).getBytes()); // This message will be delivered to consumer 10 seconds later. //设置延时等级，3 是等级 不是3s message.setDelayTimeLevel(3); // Send the message producer.send(message); } // Shutdown producer after use. producer.shutdown(); } } ScheduledMessageConsumer.java package com.example.rocketmq.demo.scheduled; 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 mon.message.MessageExt; import java.util.List; public class ScheduledMessageConsumer { public static void main(String[] args) throws Exception { // Instantiate message consumer DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ExampleConsumer"); // Subscribe topics consumer.subscribe("TopicTest", "*"); // Register message listener consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) { for (MessageExt message : messages) { // Print approximate delay time period System.out.println("Receive message[msgId=" + message.getMsgId() + "] " + (System.currentTimeMillis() - message.getStoreTimestamp()) + "ms later"); } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // Launch consumer consumer.start(); } }