RocketMQ-延时消息Demo及实现原理分析

文章目录

延时消息Producer Demo源码分析延时消息持久化内部变量含义初始化load方法start方法 延时消息调度 总结 假设有这么一个需求，用户下单后如果30分钟未支付，则该订单需要被关闭。你会怎么做？

最简单的做法，可以服务端启动个定时器，隔个几秒扫描数据库中待支付的订单，如果(当前时间-订单创建时间)>30分钟，则关闭订单。

这种方案优点是实现简单，缺点呢？

定时扫描意味着隔个几秒就得查一次数据库，频率高的情况下，如果数据库中订单总量特别大，这种高频扫描会对数据库带来一定压力，待付款订单特别多时(做个爆品秒杀活动，或者啥促销活动)，若一次性查到内存中，容易引起宕机，需要分页查询，多少也会有一定数据库层面压力

那么有没其他解决方案？关键有2点设计要求

能够在指定时间间隔后触发某个业务操作能够应对业务数据量特别大的特殊场景

RocketMQ延时消息能够完美的解决上述需求，正常的消息在投递后会立马被消费者所消费，而延时消息在投递时，需要设置指定的延时级别（不同延迟级别对应不同延迟时间），即等到特定的时间间隔后消息才会被消费者消费，这样就将数据库层面的压力转移到了MQ中，也不需要手写定时器，降低了业务复杂度，同时MQ自带削峰功能，能够很好的应对业务高峰

下面先从Demo入手，开始分析延时消息使用及原理

延时消息Producer Demo

延时消息的关键点在于Producer生产者需要给消息设置特定延时级别，消费端代码与正常消费者没有差别。

public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException { DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); //设置namesrv地址 producer.setNamesrvAddr("111.231.110.149:9876"); //启动生产者 producer.start(); //发送10条消息 for (int i = 0; i < 10; i++) { try { Message msg = new Message("TopicTest" /* Topic */, "TagA" /* Tag */, ("test message" + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */ ); //设置消息延时级别 3对应10秒后发送 msg.setDelayTimeLevel(3); SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); } catch (Exception e) { e.printStackTrace(); Thread.sleep(1000); } } /* * Shut down once the producer instance is not longer in use. */ producer.shutdown(); } }

设置消息延时级别的方法是setDelayTimeLevel()，目前RocketMQ不支持任意时间间隔的延时消息，只支持特定级别的延时消息，什么意思呢？

看下MQ中默认延时级别配置，延时级别配置代码在MessageStoreConfig#messageDelayLevel中

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

延时级别1对应延时1秒后发送消息 延时级别2对应延时5秒后发送消息 延时级别3对应延时10秒后发送消息 以此类推。。

源码分析

先从延时消息延迟级别设置与broker端消息持久化入手

延时消息持久化

通过上面Demo知道，延时消息在发送时，设置了delayLevel，两个问题

延迟级别设置后与普通消息有什么区别呢？broker接收到一个设置了延迟级别的消息后，又做了哪些特殊处理呢？

先看下Message#setDelayTimeLevel方法代码，可以看到延迟级别设置后，消息体的属性里多了一个多了一个PROPERTY_DELAY_TIME_LEVEL的属性，其值为“Delay”，value为延迟级别

public void setDelayTimeLevel(int level) { this.putProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(level)); } public static final String PROPERTY_DELAY_TIME_LEVEL = "DELAY";

再看下这个延迟属性哪里被引用了，即看下Message#getDelayTimeLevel方法在哪被引用了，可以看到在消息存储时，CommitLog中有引用

看下CommitLog中的相关代码 延时消息其核心就是两点

替换消息的topic为特定延时消息topic，queueId为delayLevel-1备份消息原有的topic，queueId，方便后面重新取出进行消息投递

CommitLog中还有一个关键点在checkMessageAndReturnSize中 延时消息的tagsCode存储的的是消息应该被投递的时间，具体后面消息调度发送会介绍

计算消息投递时间的方法computeDeliverTimestamp如下，delayLevelTable是一个key为延迟级别，value为延迟级别对应的延迟时间(毫秒)的一个map，即 消息投递时间 = 消息存储时间+延迟时间

public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) { Long time = this.delayLevelTable.get(delayLevel); if (time != null) { return time + storeTimestamp; } return storeTimestamp + 1000; }

内部变量含义

延时消息定时投递相关具体实现代码在ScheduleMessageService中，先看下变量定义

变量名含义String SCHEDULE_TOPIC = “SCHEDULE_TOPIC_XXXX”定时消息的特定topiclong FIRST_DELAY_TIME = 1000L第一次调度的时间间隔long DELAY_FOR_A_WHILE = 100L每次延时消息调度之间的时间间隔long DELAY_FOR_A_PERIOD = 10000L延时消息发送失败后，再次调度的时间间隔ConcurrentMap<Integer /* level /, Long/ delay timeMillis */> delayLevelTable延时级别做key，延时时间(毫秒)做value的map，方便从延时级别找到对应延时时间ConcurrentMap<Integer /* level /, Long/ offset */> offsetTable延时级别做key，消费进度做value的map，方便找到每个延时级别对应队列的消费进度

初始化

DefaultMessageStore在启动时，会调用ScheduleMessageService#load()方法来加载消息消费进度和初始化延迟级别对应map，然后调用ScheduleMessageService#start()方法来启动类

load方法

public boolean load() { boolean result = super.load(); result = result && this.parseDelayLevel(); return result; }

ScheduleMessageService继承自ConfigManager类，super.load()方法对应

public boolean load() { String fileName = null; try { fileName = this.configFilePath(); String jsonString = MixAll.file2String(fileName); if (null == jsonString || jsonString.length() == 0) { return this.loadBak(); } else { this.decode(jsonString); log.info("load " + fileName + " OK"); return true; } } catch (Exception e) { log.error("load " + fileName + " failed, and try to load backup file", e); return this.loadBak(); } }

分为两步

super.load()方法

一、先从指定路径configFilePath方法获取消息进度持久化存储路径，ScheduleMessageService中延时消息持久化路径存储路径代码在StorePathConfigHelper#getDelayOffsetStorePath中， 对应到物理机中路径就是${user.home}/store/config/delayOffset.json 内容如下，key为延时级别，value为消费进度

再调用decode方法将delayOffset.json转换成成map，即前面提到的offsetTable

parseDelayLevel方法 解析MessageStoreConfig中messageDelayLevel的定义，并转换成delayLevelTable

综上所述， load方法完成延时消息消费进度加载和delayLevel的加载

start方法

代码如下

步骤比较清晰

遍历延时级别map，为每个延时级别创建一个DeliverDelayedMessageTimerTask定时任务，并设置指定offset，第一次调度延时为FIRST_DELAY_TIME，即1s每隔10秒持久化延时消息消费进度

延时消息调度

延时消息定时器调度核心，在于DeliverDelayedMessageTimerTask的任务执行方法executeOnTimeup中，先看下整体结构，代码较长，步骤说明直接写在代码Todo中

其中根据，delayLevel获取消费队列id的方法如下，即queueId = delayLevel-1

public static int delayLevel2QueueId(final int delayLevel) { return delayLevel - 1; }

再具体看下找到延迟级别对应消费队列，并且从offset获取到指定消费位点消息的情况，即if (bufferCQ != null)的代码块中的内容 核心逻辑就是取出tagCode(延时消息持久化时，tagsCode存储的是消息投递时间)，解析成消息投递时间，与当前时间戳做差，判断是否应该进行消息投递，具体进行消息投递的方法，在if (countdown <= 0)中，看下代码

重新构建投递消息的关键点在于messageTimeup中，其构建了一个新的消息，并从延时消息属性中恢复出了原有的topic，queueId，再调用putMessage重新进行投递

总结

基本思路已经介绍完，梳理下延时消息实现思路

producer端设置消息delayLevel延迟级别，消息属性DELAY中存储了对应了延时级别broker端收到消息后，判断延时消息延迟级别，如果大于0，则备份消息原始topic，queueId，并将消息topic改为延时消息队列特定topic(SCHEDULE_TOPIC)，queueId改为延时级别-1mq服务端ScheduleMessageService中，为每一个延迟级别单独设置一个定时器，定时(每隔1秒)拉取对应延迟级别的消费队列根据消费偏移量offset从commitLog中解析出对应消息从消息tagsCode中解析出消息应当被投递的时间，与当前时间做比较，判断是否应该进行投递若到达了投递时间，则构建一个新的消息，并从消息属性中恢复出原始的topic，queueId，并清除消息延迟属性，从新进行消息投递