RocketMQ源码深度解析（四）延迟消息定时消息

一、延迟消息初步认知与业务价值1.1 什么是延迟消息延迟消息是指消息发送成功后消费者不会立即消费等待指定时间到达后才对消费者可见、允许消费的特殊消息类型。RocketMQ 将延迟消息分为两类架构完全不同固定级别延迟消息4.x 全量支持仅支持官方预设18个固定延迟等级无自定义时间基于队列偏移量轮询实现无时间轮任意时间定时消息5.x 新特性支持指定具体时间戳/任意延迟时长基于时间轮算法 TimerWheel TimerLog实现高精度定时调度1.2 典型落地场景订单超时关闭、未支付自动取消超时未跟进任务提醒、用户行为延时回调延迟重试、故障延后补偿任务活动定时上线、定时推送、延迟通知1.3 延迟消息核心优势替代传统JDK Timer、Scheduled线程、Redis ZSet延时队列分布式集群保障、宕机不丢失、持久化落地支持海量延时消息堆积不占用业务内存自带重试、容错、恢复机制稳定性远超业务自研定时任务二、延迟消息生产核心关注点无论是固定级别还是任意定时消息生产落地与源码学习必须关注以下核心要点2.1 可靠性关注点延迟消息是否丢失依赖 CommitLog 持久化 定时调度重试机制服务重启后延时进度是否保留基于磁盘文件位点恢复不丢失计时状态延时到期是否保证一定投递支持定时扫描、过期兜底补发2.2 性能关注点海量延时消息是否压垮Broker分级队列隔离、时间轮槽位分片、批量扫描轮询扫描是否空转浪费CPU时间轮精准调度、仅扫描当前到期槽位延时消息堆积是否影响普通消息Topic物理隔离、线程池独立调度2.3 业务规则关注点延迟消息不支持广播消费仅支持集群消费延迟消息不支持事务消息叠加固定级别延迟仅18档无法自定义任意时间5.x前痛点定时消息存在最大时长限制默认支持7天内定时三、固定级别延迟消息源码全梳理4.x核心3.1 18个固定延迟级别配置RocketMQ 4.x 通过messageDelayLevel配置固定延迟档位全局仅支持18级不可自定义任意秒数1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h级别与时间一一对应level1对应1s、level18对应2h。3.2 核心架构设计无时间轮、纯队列轮询固定级别延迟消息未使用时间轮算法采用「系统延迟Topic 分级队列 后台轮询扫描」实现架构极简、稳定性极高。系统内置TopicSCHEDULE_TOPIC_XXXX延迟消息专属主题队列映射规则QueueId DelayLevel - 1每一个延迟级别独占一个队列后台调度服务ScheduleMessageService为每一级延迟队列启动独立定时任务3.3 完整源码执行流程步骤1Producer发送延迟消息业务代码设置延迟级别消息打上延迟属性标记Message msg new Message(BUSINESS_TOPIC, 延迟消息内容.getBytes()); // 设置延迟级别 msg.setDelayTimeLevel(3); // 对应10秒延迟 rocketMQTemplate.syncSend(msg);底层会给消息添加属性PROPERTY_DELAY_TIME_LEVEL步骤2Broker拦截改写消息路由Broker 在 CommitLog 写入前拦截判断识别延迟消息后替换路由信息核心源码// CommitLog.putMessage 核心分支 if (msg.getDelayTimeLevel() 0) { // 校验延迟级别合法性 if (msg.getDelayTimeLevel() this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); } // 保存真实业务Topic、Queue到消息属性 msg.putProperty(MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic()); msg.putProperty(MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId())); // 路由替换为系统延迟Topic 对应级别队列 msg.setTopic(SCHEDULE_TOPIC); msg.setQueueId(msg.getDelayTimeLevel() - 1); }核心逻辑延迟消息不会进入业务Topic先存入系统延迟Topic实现消费者不可见达到延时效果。步骤3消息持久化与索引分发改写路由后的消息正常写入 CommitLog后台分发线程生成对应 ConsumeQueue 索引等待调度扫描。步骤4ScheduleMessageService 轮询调度Broker 启动时初始化ScheduleMessageService为18个延迟级别分别开启独立循环任务定时拉取对应延迟队列的消息对比消息投递时间与当前时间未到时间跳过等待下一轮扫描已到时间恢复原始Topic、队列信息重新投递步骤5到期消息重新投递消费到期消息剔除延迟队列恢复业务真实Topic重新写入CommitLog、生成业务索引消费者正常拉取消费延迟流程结束。3.4 固定级别延迟消息流程图调度核心代码public void executeOnTimeUp() { // 找到延迟队列对应的ConsumeQueue文件 ConsumeQueueInterface cq ScheduleMessageService.this.brokerController.getMessageStore() .getConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel)); ReferredIteratorCqUnit bufferCQ cq.iterateFrom(this.offset); long nextOffset this.offset; try { while (bufferCQ.hasNext() isStarted()) { CqUnit cqUnit bufferCQ.next(); // 获取一条ConsumeQueue记录 long offsetPy cqUnit.getPos(); int sizePy cqUnit.getSize(); long tagsCode cqUnit.getTagsCode(); // 计算下一个ConsumeQueue单元的位置下一次扫描就从这个地方开始 long currOffset nextOffset; nextOffset currOffset cqUnit.getBatchNum(); // 计算延迟是否到期 long deliverTimestamp computeDeliverTimestamp(tagsCode); long countdown deliverTimestamp - System.currentTimeMillis(); if (countdown 0) { // 还没到延迟时间 this.scheduleNextTimerTask(currOffset, DELAY_FOR_A_WHILE); ScheduleMessageService.this.updateOffset(this.delayLevel, currOffset); return; } // 获取CommitLog中的实际消息 MessageExt msgExt ScheduleMessageService.this.brokerController.getMessageStore() .lookMessageByOffset(offsetPy, sizePy); if (msgExt null) { continue; } // 处理消息转换为可投递的消息 MessageExtBrokerInner msgInner ScheduleMessageService.this.messageTimeUp(msgExt); // 时间到了就转储投递 boolean deliverSuc; if (ScheduleMessageService.this.enableAsyncDeliver) { // 异步投递默认false deliverSuc this.asyncDeliver(msgInner, msgExt.getMsgId(), currOffset, offsetPy, sizePy); } else { // K9 固定延迟级别的延迟消息同步投递 deliverSuc this.syncDeliver(msgInner, msgExt.getMsgId(), currOffset, offsetPy, sizePy); } if (!deliverSuc) { this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); return; } } } catch (Exception e) { log.error(ScheduleMessageService, messageTimeUp execute error, offset {}, nextOffset {}, this.offset, nextOffset, e); } finally { bufferCQ.release(); } // 部署下一次任务 this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); }3.5 固定级别方案优缺点优点实现简单、无复杂算法、CPU开销极低、稳定性极强、适配超大规模集群致命缺点仅支持18个固定档位不支持任意自定义时间业务灵活性极差四、RocketMQ 5.x 指定时间定时消息任意延迟4.1 方案诞生背景4.x 固定级别延迟无法满足「任意时间定时、精准延时」业务RocketMQ5.0 基于RIP-43 提案全新实现时间轮算法定时消息支持任意时间戳定时未来7天内秒级精准调度支持定时消息取消、修改海量定时消息高性能调度4.2 核心存储结构定时消息双文件定时消息放弃固定队列轮询采用TimerWheel时间轮 TimerLog日志文件双结构协同存储TimerLog顺序追加日志文件存储所有定时消息的索引信息、CommitLog偏移量、时间戳纯追加写入性能极高TimerWheel时间轮索引文件环形槽位结构按时间维度分片每一个槽位挂载对应时刻的TimerLog链表4.3 指定时间定时消息完整流程消息接收Producer指定未来具体执行时间戳发送定时消息时间槽定位Broker根据消息到期时间计算对应的时间轮Slot槽位写入TimerLog追加写入定时索引记录消息位置、时间、链表前驱指针更新时间轮指针当前槽位指向最新TimerLog记录形成时间链表时间轮驱动后台线程秒级推进时间轮指针遍历当前到期槽位解析到期消息根据槽位链表读取TimerLog定位CommitLog真实消息恢复投递到期消息正常投递至业务Topic完成定时调度五、定时调度核心时间轮算法深度详解5.1 时间轮算法核心介绍时间轮TimeWheel是一种高效定时任务调度算法核心解决传统定时任务「遍历全量任务、CPU空转、轮询低效」的痛点广泛用于中间件、网关、定时调度框架。核心思想将时间划分为固定粒度的环形槽位时间指针匀速转动仅处理当前时刻到期任务无需遍历全部任务时间复杂度 O(1)。5.2 时间轮核心组成环形槽位Slot将一段时间切分为N个时间窗口RocketMQ为秒级槽位时间指针匀速向前推进每秒移动一个槽位任务链表每个槽位挂载当前时刻需要执行的所有定时任务底层存储RocketMQ落地为TimerWheel磁盘文件支持持久化不丢失5.3 RocketMQ 原生磁盘时间轮完整实现原理源码级普通内存时间轮仅适用于单机临时定时任务重启丢失、无法支撑海量堆积。RocketMQ 5.x 基于RIP-43实现了磁盘持久化环形时间轮完全脱离内存限制、支持重启恢复、支持7天超长定时、百万级消息调度是中间件领域独有的工业级时间轮落地方案。5.3.1 核心硬件与文件参数固定不可改RocketMQ 时间轮采用固定精度、固定窗口、定长结构体设计极致适配磁盘顺序读写时间精度1秒/槽秒级精准调度时间窗口默认覆盖7天总槽位数量7×24×3600个 SlotTimerWheel 文件大小固定约 37MB极小内存占用、常驻磁盘调度模式环形复用、指针步进、过期批量触发5.3.2 Slot 槽位定长存储结构核心数据结构每个时间槽位 Slot 固定占用 32Byte结构化存储当前秒所有定时消息的链表头尾指针是时间轮与 TimerLog 关联的核心桥梁字段定义如下// 单Slot 32字节固定结构 delayed_time(8Byte) // 当前槽位对应的绝对时间戳 first_pos(8Byte) // 当前秒任务链表的TimerLog起始偏移量 last_pos(8Byte) // 当前秒任务链表的TimerLog末尾偏移量 msg_count(4Byte) // 当前秒堆积的定时消息数量 reserved(4Byte) // 预留填充位对齐磁盘定长读写核心设计槽位不存消息本体只存链表指针所有消息实体索引全部存在顺序写的 TimerLog 文件中实现「时间轮索引日志存储」的读写分离架构。5.3.3 TimerLog 日志结构与链表挂载机制TimerLog 是纯顺序追加写入的磁盘日志文件每条定时消息写入都会生成一条定长日志记录每条记录携带前驱指针形成单秒内双向链表结构。单条 TimerLog 核心字段commitlog_pos真实消息在 CommitLog 中的物理偏移量精准定位原消息prev_pos同秒上一条消息的 TimerLog 偏移量链表前驱next_pos同秒下一条消息的 TimerLog 偏移量链表后继delayed_time消息到期时间戳挂载逻辑同一秒到期的多条消息通过 prev/next 指针串联成链表时间轮 Slot 只记录该链表首尾位置无需遍历全量消息极致提升读写效率。5.3.4 定时消息入队完整实现流程源码链路时间戳校验Broker 拦截定时消息校验到期时间在「当前时间~7天后」区间超出直接报错槽位计算通过到期时间对总槽位数取模精准定位对应 Slot 位置slotIndex delayedTime % 1209600顺序写TimerLog追加写入当前定时消息索引记录前驱指针更新Slot指针更新对应槽位的 last_pos、msg_count完成链表尾部挂载落盘持久化TimerLog 顺序刷盘、TimerWheel 索引更新保证断电不丢失任务关系5.3.5 时间轮指针步进与出队执行原理RocketMQ 后台专属时间轮线程TimerDequeueService每秒驱动指针前进一格实现 O(1) 调度指针指向当前系统时间对应的 Slot 槽位判断当前槽位 msg_count 0判定是否存在到期任务通过 first_pos、last_pos 遍历当前秒 TimerLog 链表根据 commitlog_pos 读取 CommitLog 原始定时消息恢复消息真实业务 Topic、队列信息重新投递至业务队列清空当前槽位任务计数与指针完成本轮调度5.3.6 7天环形复用与超长定时实现针对超过7天的定时任务RocketMQ 采用时间轮环形复用轮回计数实现通过时间戳差值计算需要轮转的圈数指针每走完一轮7天窗口自动叠加轮回计数未到期任务保留在原槽位等待下一轮指针命中无需扩容文件、无需迁移数据完美支持最长7天定时5.3.7 重启断点续跑机制高可靠核心时间轮所有状态持久化在磁盘文件Broker 重启后自动恢复现场加载本地 TimerWheel 文件读取最后一次调度的时间指针位置从断点位置继续向后步进不会重复调度历史任务扫描当前所有未到期槽位重建内存链表映射恢复正常秒级步进调度保证定时任务不丢、不重复、不遗漏5.3.8 核心源码片段槽位寻址与任务挂载// 根据到期时间计算时间轮槽位下标 public int getSlotIndex(long delayedTime) { // 总槽位7天秒级总数 1209600 return (int) (delayedTime % TIMER_WHEEL_TOTAL_SLOT); } // 新定时消息挂载至时间轮链表尾部 public void appendTimerLog(TimerLogRecord record, long delayedTime) { int slotIdx getSlotIndex(delayedTime); Slot slot this.slotTable[slotIdx]; // 1. 写入TimerLog顺序日志 long logOffset timerLog.append(record); // 2. 更新槽位链表首尾指针 if (slot.getMsgCount() 0) { slot.setFirstPos(logOffset); } slot.setLastPos(logOffset); slot.incrementCount(); // 3. 刷盘持久化索引 flushWheelAndLog(); }5.3.9 RocketMQ磁盘时间轮四大工程优势零CPU空转仅处理当前秒到期槽位无全局遍历、无无效轮询磁盘级海量支撑完全落地磁盘百万级定时消息无内存压力精准秒级调度固定1秒粒度无档位限制适配任意定时场景崩溃自愈全量持久化、断点续跑重启不丢失定时任务状态5.43 工作原理通俗拆解初始化固定长度的环形时间轮默认覆盖7天时间窗口新定时消息根据到期时间哈希取模定位对应槽位任务挂载至对应槽位的链表尾部后台线程每秒驱动指针前进一格指针指向的当前槽位批量执行所有到期任务超长定时任务自动轮转复用时间轮窗口实现超长定时支持5.5 RocketMQ时间轮优势对比轮询精准调度秒级精度无固定档位限制性能极致仅扫描当前到期槽位无空轮询CPU损耗海量支撑槽位分片、链表挂载支持百万级定时消息堆积持久化可靠磁盘存储时间轮状态重启不丢失定时进度六、双模式节点关系与联动机制6.1 固定级别延迟消息节点关系Producer → Broker路由拦截 → 系统延迟Topic → 分级Queue → Schedule定时扫描 → 业务Topic → Consumer核心队列与延迟级别强绑定每个队列独立调度节点职责单一、完全隔离。6.2 任意定时消息节点关系Producer指定时间 → TimerWheel槽位映射 → TimerLog持久化 → 时间轮指针驱动 → 到期解析 → 重新投递消费核心时间维度分片存储脱离队列限制以时间为核心调度。6.3 两套机制核心区别对比维度固定级别延迟消息(4.x)任意时间定时消息(5.x)核心算法队列轮询扫描时间轮调度算法时间精度固定18档精度粗糙秒级精准、任意时间存储结构系统延迟Topic分级队列TimerWheel TimerLog性能损耗低定时轮询无复杂计算极低精准槽位触发业务灵活性差无法自定义时间极高适配所有定时场景七、源码核心关键点总结面试绝杀固定延迟核心消息写入时动态替换Topic为系统延迟Topic按级别绑定队列后台定时轮询到期补发无时间轮参与路由还原关键点延迟消息必须缓存原始Topic和QueueID到期后恢复路由才能被业务消费者消费定时消息核心5.x通过时间轮槽位映射TimerLog链表实现任意时间调度解决固定档位痛点时间轮核心关键点时间分片、槽位挂载、指针驱动、O(1)调度、无空轮询损耗可靠性关键点双方案均基于磁盘持久化重启位点恢复保证延时消息不丢失、不重复、不遗漏性能关键点级别隔离、时间分片、任务链表挂载海量延时消息互不影响八、全篇核心总结1、RocketMQ 延迟消息分为4.x固定级别延迟队列轮询与5.x任意时间定时时间轮算法两套完全独立的实现架构。2、固定延迟核心原理路由替换分级队列后台轮询扫描架构简单稳定唯一短板是时间不灵活。3、定时消息核心原理TimerWheel时间轮分片TimerLog顺序日志秒级精准调度支持任意自定义延迟时间。4、时间轮算法是高性能定时调度的核心通过时间槽位拆分、指针驱动彻底解决传统轮询CPU空转、全量遍历的性能问题。5、两套方案均实现延时消息消费者不可见、到期自动投递的核心能力且基于磁盘持久化保障分布式场景高可靠。