RocketMQ延迟消息是如何实现的?

RocketMQ的延迟消息实现机制非常巧妙，其核心是通过多级时间轮 + 定时任务 + 消息重投递来实现的。以下是详细实现原理：

⏰ 一、延迟消息的核心设计

预设延迟级别（非任意时间）
RocketMQ不支持任意时间延迟，而是预设了18个固定延迟级别（1-18），每个级别对应固定延迟时间：
```
// 源码中的延迟级别定义 (MessageStoreConfig类)
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
```
延迟消息处理流程：

🔧 二、Broker端实现细节

1. 特殊主题存储

所有延迟消息先存入内部主题：SCHEDULE_TOPIC_XXXX
该主题包含 18个队列，每个队列对应一个延迟级别

消息结构包含关键元数据：

class Message {private String topic;      // 原始主题（如ORDER_TOPIC）private int delayLevel;   // 延迟级别（3=10秒）private long storeTimestamp; // 存储时间戳// ...其他字段
}

2. 时间轮调度器（核心）

public class ScheduleMessageService extends ConfigManager {// 延迟级别对应的Timerprivate final ConcurrentMap<Integer, Timer> timerTable = new ConcurrentHashMap<>(32);// 延迟级别对应的处理队列private final ConcurrentMap<Integer, Long> offsetTable =new ConcurrentHashMap<>(32);
}

每个延迟级别独立Timer：为18个级别分别创建定时器
时间轮算法：使用HashedWheelTimer高效管理延迟任务

3. 消息重投递过程

当延迟时间到达时：

从SCHEDULE_TOPIC_XXXX的对应队列拉取消息
清除消息的delayLevel属性
将消息写入原始目标Topic
消费者此时可正常消费

⚡ 三、源码级执行流程

消息接收（Broker端）：

// DefaultMessageStore.putMessage()
if (msg.getDelayTimeLevel() > 0) {// 修改Topic为SCHEDULE_TOPIC_XXXXtopic = ScheduleMessageService.SCHEDULE_TOPIC;// 计算目标队列：queueId = delayLevel - 1queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
}

定时扫描（每秒执行）：

// ScheduleMessageService.executeOnTimeup()
for (int level = 1; level <= 18; level++) {long delayTimeMillis = computeDeliverTimestamp(level, storeTimestamp);if (now >= delayTimeMillis) {// 触发重投递deliverDelayedMessage(level);}
}

重投递关键操作：

MessageExt msgExt = scheduleMessageIterator.next();
// 恢复原始Topic/Queue
MessageExtBrokerInner msgInner = rebuildMessage(msgExt);
// 存入CommitLog（真实Topic）
PutMessageResult result = defaultMessageStore.putMessage(msgInner);

📊 四、延迟级别与时间映射

延迟级别	实际延迟时间	对应队列ID
1	1秒	queue0
2	5秒	queue1
3	10秒	queue2
4	30秒	queue3
5	1分钟	queue4
…	…	…
18	2小时	queue17

⚠️ 五、使用注意事项

不支持任意时间延迟
只能选择预设的18个级别（可通过修改配置扩展级别）
最大延迟时间限制
默认最大2小时，修改需调整配置并重启Broker
精度误差
实际延迟可能有1-2秒误差（受扫描周期影响）
资源消耗
高并发延迟消息会显著增加Broker的CPU负载

🔄 六、生产环境优化建议

调整扫描频率（平衡精度与CPU）

# broker.conf
flushDelayOffsetInterval=1000  # 默认1秒，可调大到3秒

扩展延迟级别
修改messageDelayLevel配置增加自定义级别：
```
messageDelayLevel=1s 5s 10s 30s 1m 2m 5m 10m 30m 1h 2h 6h 12h
```
监控关键指标
- ScheduleMessageService_* 开头的指标
- 延迟队列积压情况（通过Admin CLI查看）