009 rocketmq延时消息

延迟消息

两个关键要素：临时存储 延时服务
内置队列，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个步骤进⾏详细的讲解：

1. 修改消息Topic名称和队列信息
2. 转发消息到延迟主题SCHEDULE_TOPIC_XXXX的CosumeQueue中
3. 延迟服务消费SCHEDULE_TOPIC_XXXX消息
4. 将信息重新存储到CommitLog中
5. 将消息投递到⽬标Topic中
6. 消费者消费⽬标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.common.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.common.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();
    }
}