什么是延迟队列

指消息发送到某个队列后，在指定多长时间之后才能被消费。
消费消息，按照0到18级别来，0 表示不延迟，1表示延迟1s，大于等于18表示延迟2h
按照级别一次类推
默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m8m 9m 10m 20m 30m 1h 2h”，18个level

测试

生产端

test-topic 有四个队列，测试代码将0-99 分别放到不同的队列中，根据不同的队列设置不同的延时值(0(0s)，4(30s)，8(4m)，12(8m))

@GetMapping("/send/{message}")
public void send(@PathVariable("message") String messageStr) {
    rocketMQTemplate.setMessageQueueSelector((list, message, o) -> {
        int index = Integer.valueOf((String) o) % 4;
        message.setDelayTimeLevel(index * 4);
        return list.get(index);
    });
    for (int i = 0; i < 100; i++) {
        Message message = MessageBuilder
        .withPayload("message is :" + i)
        .setHeader("avg", i)
        .build();
        rocketMQTemplate.sendOneWayOrderly("test-topic", message, String.valueOf(i));
    }
}

消费端

可以看到，队列1的消息比队列0的消息慢了30s

原理

rocketmq会创建一个专门处理延迟消息的topic叫做SCHEDULE_TOPIC_XXXX，这个topic中会根据延迟级别创建队列，然后将不同延迟级别的消息放到指定的队列，保证顺序消费，否则如果高延迟和低延迟的放一起的话，高延迟消息必然会影响低延迟消息的消费，等待指定时间后，SCHEDULE_TOPIC_XXXX会把消息发送到真正的topic

(1)、修改消息Topic名称和队列信息
RocketMQ Broker端在存储生产者写入的消息时，首先都会将其写入到CommitLog中。之后根据消息中的Topic信息和队列信息，将其转发到目标Topic的指定队列(ConsumeQueue)中。

由于消息一旦存储到ConsumeQueue中，消费者就能消费到，而延迟消息不能被立即消费，所以这里将Topic的名称修改为SCHEDULE_TOPIC_XXXX，并根据延迟级别确定要投递到哪个队列下。同时，还会将消息原来要发送到的目标Topic和队列信息存储到消息的属性中。

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

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

需要注意的是，会将计算出的投递时间当做消息Tag的哈希值存储到CosumeQueue中，CosumeQueue单个存储单元组成结构如下图所示：

其中：

Commit Log Offset：记录在CommitLog中的位置；
Size：记录消息的大小；
Message Tag HashCode：记录消息Tag的哈希值，用于消息过滤。特别的，对于延迟消息，这个字段记录的是消息的投递时间戳。这也是为什么java中hashCode方法返回一个int型，只占用4个字节，而这里Message Tag HashCode字段却设计成8个字节的原因；
(3)、延迟服务消费SCHEDULE_TOPIC_XXXX消息
Broker内部有一个ScheduleMessageService类，其充当延迟服务，主要是消费SCHEDULE_TOPIC_XXXX中的消息，并投递到目标Topic中。

ScheduleMessageService在启动时，其会创建一个定时器Timer，并根据延迟级别的个数，启动对应数量的TimerTask，每个TimerTask负责一个延迟级别的消费与投递。

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

(4)、将信息重新存储到CommitLog中
在将消息到期后，需要投递到目标Topic。由于在第一步已经记录了原来的Topic和队列信息，因此这里重新设置，再存储到CommitLog即可。此外，由于之前Message Tag HashCode字段存储的是消息的投递时间，这里需要重新计算tag的哈希值后再存储。

(5)、将消息投递到目标Topic中
这一步与第二步类似，不过由于消息的Topic名称已经改为了目标Topic。因此消息会直接投递到目标Topic的ConsumeQueue中，之后消费者即消费到这条消息。

(6)、消费者消费目标topic中的数据。

RocketMQ延时队列

什么是延迟队列

测试

生产端

消费端

原理