RocketMQ 4.X升级5.X定时/延时消息实战
前言：
业务场景中需要实现对某些记录进行检查，给定一小时的时效，过期未检查的记录需要修改状态为失效，所以想到了用MQ的延迟消息（自我生产消费）去实现，但一开始使用的是4.2版本的延时消息，发现给定的一小时延迟往往不够准确，深入研究后发现4.2版本的延时消息实现有局限（4.X版本依赖轮询机制，当延时队列消息积压时，定时线程处理延迟，会导致实际消费时间与预期存在较大误差），所以升级为5.1版本（更加高效的时间轮算法）的定时消息。
对应4.2版本的延迟消息生产者代码
/**
* 发送延迟消息（4.x用法）
*
* @param topic 消息主题
* @param content 消息体内容
* @param delayLevel 延迟级别 (1-18)
*/
public void sendDelayMessage(String topic, Object content, int delayLevel) {
Message<Object> message = MessageBuilder.withPayload(content).build();
// 发送消息，并指定延迟级别
rocketMQTemplate.syncSend(topic, message, 3000, delayLevel);
}
这里生产者发送延迟消息 给定的delayLevel值是17 对应 1小时 实际运行中发现延迟时间往往无法精确到1小时
查看4.2版本RocketMQ源码 该版本延迟消息的实现 其实是放在同一个topic中的
可以看到当延迟等级DelayTimeLevel 有值得时候 topic被覆盖为了SCHEDULE_TOPIC_XXXX 正是这个动作带来了后续消息积压导致延迟时间不准确
更换5.X版本的定时消息（4.X版本的延迟消息新版本中做了保留，准确讲4.X叫延迟消息，5.X叫定时消息）
/**
* 发送延迟消息 (兼容旧版 delayLevel 的 RocketMQ 5.x 方式)
*
* @param topic 消息主题
* @param content 消息体内容
* @param delayLevel 延迟级别 (1-18)
*/
public void sendDelayMessage(String topic, Object content, int delayLevel) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
try {
// 将旧的 delayLevel 映射为具体的毫秒数
long deliverTimeMs = this.mapDelayLevelToMs(delayLevel);
// 将 content 对象序列化为 JSON 字符串 这是最关键的一步！确保消费端RocketMQListener能正确解析
String jsonContent = JSON.toJSONString(content);
// 创建 RocketMQ 原生 Message 对象
Message rocketMsg = new Message(topic, jsonContent.getBytes(StandardCharsets.UTF_8));
// 设置延迟时间 使用时间戳 固定到未来的某个时间点精确发送，精度的误差为消费端的实际消费耗时，毫秒级
rocketMsg.setDeliverTimeMs(System.currentTimeMillis() + deliverTimeMs);// 这特么是绝对时间 需要+当前时间
SendResult sendResult = rocketMQTemplate.getProducer().send(rocketMsg);
logger.info("延迟消息发送成功 (5.x Native)，主题: {}, 是否成功：{}", topic, sendResult.getSendStatus());
} catch (Exception e) {
logger.error("延迟消息发送失败，主题: {}", topic, e);
throw e;
}
}
这里为了兼容4.X版本 对delayLevel做了毫秒值的转化，实现对相关业务代码的低风险改造 其中最重要的是setDeliverTimeMs方法（这里容易用错成setDelayTimeMs方法需要注意区别！），设置的是绝对时间即当前时间往后延相应的毫秒值
查看5.1.1版本源码 找到setDelayTimeMs和setDeliverTimeMs的区别，首先查看源码中的示例
源码示例中的描述这两个方法效果一致 但其实进一步查看处理这两个参数的方法就能发现差异
setDelayTimeMs是到了MQ之后被加上了系统当前时间的，MQ的系统当前时间 ！！！这里就产生了一定的延迟了，对精确要求高还是使用setDelayTimeMs更稳妥点
关于5.1.4版本定时消息的具体实现 查看源码
checkIfTimerMessage方法内判断是否有TIMER_DELIVER_MS、、PROPERTY_TIMER_DELAY_MS、TIMER_DELAY_SEC即生产者设置的setDeliverTimeMs、setDelayTimeMs、setDelayTimeSec
查看transformTimerMessage方法 就是上文中我们查看setDelayTimeMs和setDelayTimeMs差异的地方
可以看到与4.2版本类似 消息交由统一的topic管理
rmq_sys_wheel_timer 这个topic就是5.X版本里开始出现的定时消息了，相关的就是时间轮算法逻辑，源码中处理这个定时消息的逻辑在 TimerMessageStore中，感兴趣可以自行了解下其中具体的时间轮算法的实现，简单一句话总结下，时间轮算法的轮询线程也只需检查当前时间刻度对应的任务队列，而非所有任务。
附 将 RocketMQ 4.x 的延迟级别映射为毫秒数的方法
/**
* 将 RocketMQ 4.x 的延迟级别映射为毫秒数
* @param delayLevel 延迟级别 (1-18)
* @return 延迟毫秒数
*/
private long mapDelayLevelToMs(int delayLevel) {
if (delayLevel < 1 || delayLevel > 18) {
throw new IllegalArgumentException("delayLevel must be between 1 and 18");
}
String timeStr = DELAY_TIME_ARRAY[delayLevel – 1];
// 简单的解析逻辑
if (timeStr.endsWith("s")) {
return Long.parseLong(timeStr.substring(0, timeStr.length() – 1)) * 1000;
} else if (timeStr.endsWith("m")) {
return Long.parseLong(timeStr.substring(0, timeStr.length() – 1)) * 60 * 1000;
} else if (timeStr.endsWith("h")) {
return Long.parseLong(timeStr.substring(0, timeStr.length() – 1)) * 60 * 60 * 1000;
}
return 0;
}
写在最后：
生产实际查看数据更新间隔 有1S的误差，因为消息生产发送到轮询到消息进行消费以及数据库更新的时间大概平均就是1S，如果有更加精确的时效要求，可以取一定量的生产实际情况取均值做偏移即可。
