🍊 Java学习：Java从入门到精通总结

🍊 深入浅出RocketMQ设计思想：深入浅出RocketMQ设计思想

🍊 绝对不一样的职场干货：大厂最佳实践经验指南


📆 最近更新：2022年8月5日

🍊 个人简介：通信工程本硕💪、Java程序员🌕。做过科研paper，发过专利，优秀的程序员不应该只是CRUD

🍊 点赞 👍 收藏 ⭐留言 📝 都是我最大的动力！

---

定义

延迟消息： 生产者发送消息之后，不能立刻被消费者消费，而是应该等到一定时间之后才可以被消费。

例如： 在淘宝购物时，当用户提交一个订单之后，如果30min还没有支付的话，就给用户发送一条提示消息

Broker提供延迟消息功能的核心思路是一致的：将延迟消息先存放在一个临时的空间里，等到期后才发送给Topic

如上图所示，实现延迟消息的步骤如下：

1. Producer发送一个延迟消息
2. Broker将其通过临时存储进行存储
3. 延迟服务（delay service）检查消息持否到期，将到期的消息投递到目标Topic里
4. 消费者消费延迟消息

根据上面的流程不难发现，临时存储和延迟服务是实现延迟消息的关键，并且外部的Producer和Consumer是感知不到的

RocketMQ将延迟消息存储在一个内部主题SCHEDULE_TOPIC_XXXX里，只支持固定的时间精度，如5s、10s、1min等

Kafka原生不支持延迟消息，此时就可以借助代理 + 第三方存储来实现延迟消息

如果要借助第三方存储服务，则在技术选型时需要满足一下几个条件：
1. 高性能： 写入延迟要低，MQ的一个重要作用是削峰填谷，所以在高峰期写入性能一定要高，MySQL就不满足

2. 高可靠： 延迟消息写入之后，不能丢失

3. 支持排序： 要支持按照时间先后来排序，例如先发一条10s的延迟消息A，再发一条5s的延迟消息B，B要比A先消费掉

DDMQ

DDMQ就使用了代理 + 第三方存储来实现延迟消息：

上图的第三方存储技术选型是rocksdb，它是一个高性能KV存储（类似于Redis），并且支持排序

执行步骤：

1. 生产者将延迟消息发送给代理，代理将其发送到一个缓冲Topic中
2. 消息存储在rocksdb里，以时间为key
3. delay service判断消息时间到期后就将其发送到目标Topic里
4. 消费者消费Topic里的数据

不难发现，这种设计体现了解耦的思路：

• delay service的延迟投递能力是独立于Broker实现的，无序对Broker做任何改造，所以理论上对于任意MQ类型都可以提供延迟消息能力

delay service的设计细节

• 高可用： delay service以分布式部署了多个节点
• 最终一致性： 每个delay service都要消费缓冲Topic的全量数据，这样就有了多个备份，可以保证数据不丢失
• 避免重复消费： delay service选择一个主节点Master将消息投递到目标Topic里，这一步可以借助zookeeper来实现
• Master需要将自己当前已经投递到的时间记录到一个共享存储里，如果Master挂了，则会重新选举出一个新的Master，从之前的时间点继续开始投递

RocketMQ中的延迟消息

RocketMQ默认只支持18个时间延迟消息，由配置项messageDelayLevel决定：

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

Broker每个节点在启动时都会创建18个级别的队列，如果想要修改，直接在安装了RocketMQ的机器上修改配置文件即可。

生产者发送延迟消息

只需要设置一下延迟级别，延迟级别对应上面messageDelayLevel的索引：

Message msg = new Message();
msg.setTopic("Topic A");
msg.setTags("Tag A");
msg.setBody("delay message".getBytes());
// 设置延迟级别 = 1，对应延迟1s
msg.setDelayTimeLevel(1);
1
2
3
4
5
6

如果设置的延迟级别超过最大值，则会自动设置成最大值

Broker端处理延迟消息

Broker端处理延迟消息一共分为5个步骤，如下图所示：

整个过程分为5个步骤：

1. 生产者发送延迟消息到Broker里
2. 把消息转发到SCHEDULE_TOPIC_XXXX主题下的队列中
3. 延迟服务定期消费SCHEDULE_TOPIC_XXXX主题下的消息，到时间了就把它拿到CommitLog中
4. 消息重新被投放到目标Topic里
5. 消费者消费延迟消息

---

1. 生产者发送延迟消息到Broker里

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {

    //事务消息处理
    if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        // 如果是延迟消息
        if (msg.getDelayTimeLevel() > 0) {
            // 如果设置的值过大，则设置为最大延迟级别
            if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
            }

            // 修改Topic
            topic = ScheduleMessageService.SCHEDULE_TOPIC;
            // 根据延迟级别，决定要将其投递到那个队列中
            queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

            // 记录原始的 topic 和 队列信息
            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()));

            // 修改topic和队列信息
            msg.setTopic(topic);
            msg.setQueueId(queueId);
        }
    }

    // ......

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

可以看到，RocketMQ的Broker端在存储生产者写入消息时，首先将其写入CommitLog里，为了不让用户立刻就能消费到这条消息，这里先将Topic的名称修改为SCHEDULE_TOPIC_XXXX，并且根据设置的延迟级别选择将消息投放到哪一个队列里。

在修改Topic的名称时，还会将原来要发送到的Topic和队列信息存储在消息的属性里

2. 把消息转发到SCHEDULE_TOPIC_XXXX主题下的队列中

这一步的操作是异步执行的，在转发过程中会对延迟消息进行特殊处理，主要是计算这条消息什么时间要被投递到最终的Topic里

投递时间 = 存储消息的世界 + 延迟时间

ConsumeQueue的单个存储结构如图所示：

• Commit Log Offset：消息在Commit中的存放位置
• Size：消息体大小
• Message Tag HashCode：消息Tag的hash值记录在此。对于延迟消息，这里记录的是消息的投递时间，所以才是8Byte而不是4Byte

public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC, final boolean readBody) {
    	// ......
        try {
        	// ......
                {	// 获取延迟级别
                    String t = propertiesMap.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
                    // 如果消息需要投递到`SCHEDULE_TOPIC_XXXX`主题里
                    if (ScheduleMessageService.SCHEDULE_TOPIC.equals(topic) && t != null) {
                        int delayLevel = Integer.parseInt(t);
                        // 延迟级别大于0，则计算目标投递时间，并将其记录在Message Tag HashCode里
                        if (delayLevel > 0) {
                            tagsCode = this.defaultMessageStore.getScheduleMessageService().computeDeliverTimestamp(delayLevel,
                                    storeTimestamp);
                        }
                    }
                }
            }
            // ....
            return new DispatchRequest(//
                topic, // 1
                queueId, // 2
                physicOffset, // 3
                totalSize, // 4
                tagsCode, // 5
                storeTimestamp, // 6
                queueOffset, // 7
                keys, // 8
                uniqKey, //9
                sysFlag, // 9
                preparedTransactionOffset// 10
            );
        } catch (Exception e) {
        }

        return new DispatchRequest(-1, false /* success */);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

3. 延迟服务定期消费SCHEDULE_TOPIC_XXXX主题下的消息，到时间了就把它拿到CommitLog中

Broker内部的ScheduleMessageService类提供延迟服务，该服务消费SCHEDULE_TOPIC_XXXX主题下的消息，并投递到目标Topic下。

public void start() {
        if (started.compareAndSet(false, true)) {
        	// 创建定时器
            this.timer = new Timer("ScheduleMessageTimerThread", true);
            // 循环所有的延迟级别，delayLevelTable记录每个延迟级别对应的延迟时间
            for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
                Integer level = entry.getKey();
                Long timeDelay = entry.getValue();
                Long offset = this.offsetTable.get(level);
                if (null == offset) {
                    offset = 0L;
                }
				// 针对每个延迟级别创建一个TimerTask
                if (timeDelay != null) {
                    this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
                }
            }

            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());
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

可以看出，ScheduleMessageService启动时会启动一个Timer，并根据延迟级别的个数创建对应数量的TimerTask，每个TimerTask负责一个延迟级别中的消息的投递。

每个TimerTask在检查消息是否到期时，首先会检查对应队列中尚未投递的第一条消息，如果这条消息没到期，说明后面的消息都没到期，就不用再做进一步的检查了。

在第1步执行时，已经存下了原先要投递到的目标TOPIC和队列信息，这里只需要重新设置一下即可。此外，在第2步的执行过程里Message Tag HashCode存储的是消息的投递时间，这里要重新计算成生产者设置的Tag的hash值。

4. 消息重新被投放到目标Topic里
与第2步类似，只不过这里的Topic已经改成了目标Topic，此时消费者就可以消费到这条消息

延迟消息与消息重试机制的关系

RocketMQ提供了消息消费失败重试的机制，当消息被消费失败时，如果返回的值是RECONSUME_LATER的话就会重试：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();

consumer.registerMessageListener(new MessageListenerConcurrently() {
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});
1
2
3
4
5
6
7

默认的重试次数是16次，并且会随着失败次数的增加，重试的时间间隔也会增加，具体如下表：

实际上，消息重试的时间间隔，就对应延迟消息的后16个级别：

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

原因是消息重试的底层也是借助延迟消息来实现的，在消费失败的情况下会重新把消息当做延迟消息投递回去，官方文档也给了说明：