延时消息是指发送到 RocketMQ 后不会马上被消费者拉取到而是等待固定的时间才能被消费者拉取到。延时消息的使用场景很多比如电商场景下关闭超时未支付的订单某些场景下需要在固定时间后发送提示消息。1 生产者首先看一个生产者发送延时消息的官方示例代码public static void main(String[] args) throws Exception { // Instantiate a producer to send scheduled messages DefaultMQProducer producer new DefaultMQProducer(ExampleProducerGroup); // Launch producer producer.start(); int totalMessagesToSend 100; for (int i 0; i totalMessagesToSend; i) { Message message new Message(TestTopic, (Hello scheduled message i).getBytes()); // This message will be delivered to consumer 10 seconds later. message.setDelayTimeLevel(3); // Send the message producer.send(message); } // Shutdown producer after use. producer.shutdown(); }从上面的代码可以看到跟普通消息不一样的是消息设置 setDelayTimeLevel 属性值这里设置为 3这里最终将 3 这个延时级别复制给了 DELAY 属性。关于延时级别可以看下面这个定义//MessageStoreConfig类 private String messageDelayLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h;这里延时级别有 18 个上面的示例代码中延迟级别是 3消息会延迟 10s 后消费者才能拉取。2 Broker 处理2.1 写入消息Broker 收到消息后会将消息写入 CommitLog。在写入时会判断消息 DELAY 属性是否大于 0。代码如下//CommitLog 类 if (msg.getDelayTimeLevel() 0) { if (msg.getDelayTimeLevel() this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); } topic TopicValidator.RMQ_SYS_SCHEDULE_TOPIC; int queueId ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()); // Backup real topic, queueId 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())); msg.setTopic(topic); msg.setQueueId(queueId); }从上面的代码可以看到CommitLog 写入时并没有直接写入而是把 Topic 改为 SCHEDULE_TOPIC_XXXX把 queueId 改为延时级别减 1。因为延时级别有 18 个所以这里有 18 个队列。如下图2.2 调度消息延时消息写入后会有一个调度任务不停地拉取这些延时消息这个逻辑在类 ScheduleMessageService。这个类的初始化代码如下public void start() { if (started.compareAndSet(false, true)) { this.load(); this.deliverExecutorService new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl(ScheduleMessageTimerThread_)); //省略部分逻辑 for (Map.EntryInteger, Long entry : this.delayLevelTable.entrySet()) { Integer level entry.getKey(); Long timeDelay entry.getValue(); Long offset this.offsetTable.get(level); if (null offset) { offset 0L; } if (timeDelay ! null) { //省略部分逻辑 this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS); } } //省略持久化的逻辑 } }上面的 load() 方法会加载一个 delayLevelTableConcurrentHashMap类型key 保存延时级别从 1 开始value 保存延时时间(单位是 ms)。load() 方法结束后创建了一个有 18 个核心线程的定时线程池然后遍历 delayLevelTable创建 18 个任务DeliverDelayedMessageTimerTask进行每个延时级别的任务调度。任务调度的代码逻辑如下public void executeOnTimeup() { ConsumeQueue cq ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel)); if (cq null) { this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_WHILE); return; } SelectMappedBufferResult bufferCQ cq.getIndexBuffer(this.offset); if (bufferCQ null) { //省略部分逻辑 this.scheduleNextTimerTask(resetOffset, DELAY_FOR_A_WHILE); return; } long nextOffset this.offset; try { int i 0; ConsumeQueueExt.CqExtUnit cqExtUnit new ConsumeQueueExt.CqExtUnit(); for (; i bufferCQ.getSize() isStarted(); i ConsumeQueue.CQ_STORE_UNIT_SIZE) { long offsetPy bufferCQ.getByteBuffer().getLong(); int sizePy bufferCQ.getByteBuffer().getInt(); long tagsCode bufferCQ.getByteBuffer().getLong(); //省略部分逻辑 long now System.currentTimeMillis(); long deliverTimestamp this.correctDeliverTimestamp(now, tagsCode); nextOffset offset (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); long countdown deliverTimestamp - now; if (countdown 0) { this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); return; } MessageExt msgExt ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy); if (msgExt null) { continue; } MessageExtBrokerInner msgInner ScheduleMessageService.this.messageTimeup(msgExt); //事务消息判断省略 boolean deliverSuc; //只保留同步 deliverSuc this.syncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy); if (!deliverSuc) { this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); return; } } nextOffset this.offset (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); } catch (Exception e) { log.error(ScheduleMessageService, messageTimeup execute error, offset {}, nextOffset, e); } finally { bufferCQ.release(); } this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); }这段代码可以参考下面的流程图来进行理解上面有一个修正投递时间的函数这个函数的意义是如果已经过了投递时间那么立即投递。代码如下private long correctDeliverTimestamp(final long now, final long deliverTimestamp) { long result deliverTimestamp; long maxTimestamp now ScheduleMessageService.this.delayLevelTable.get(this.delayLevel); if (deliverTimestamp maxTimestamp) { result now; } return result; }注意消息从 CommitLog 转发到 ConsumeQueue 时会判断是否是延时消息Topic SCHEDULE_TOPIC_XXXX 并且延时级别大于 0如果是延时消息就会修改 tagsCode 值为消息投递的时间戳而 tagsCode 原值是 tag 的 HashCode。代码如下//CommitLog类checkMessageAndReturnSize方法 if (delayLevel 0) { tagsCode this.defaultMessageStore.getScheduleMessageService().computeDeliverTimestamp(delayLevel, storeTimestamp); }如下图而 ScheduleMessageService 调度线程将消息从 ConsumeQueue 重新投递到原始队列中时会把 tagsCode 再次修改为 tag 的 HashCode代码如下//类MessageExtBrokerInner这个方法被 messageTimeup 方法调用。 public static long tagsString2tagsCode(final TopicFilterType filter, final String tags) { if (null tags || tags.length() 0) { return 0; } return tags.hashCode(); }如下图2.3 一个问题如果有一个业务场景要求延时消息 3 小时才能消费而 RocketMQ 的延时消息最大延时级别只支持延时 2 小时怎么处理这里提供两个思路供大家参考在 Broker 上修改 messageDelayLevel 的默认配置在客户端缓存 msgId先设置延时级别是 182h当客户端拉取到消息后首先判断有没有缓存如果有缓存则再次发送延时消息这次延时级别是 171h如果没有缓存则进行消费。3 总结经过上面的讲解延时消息的处理流程如下最后延时消息的延时时间并不精确这个时间是 Broker 调度线程把消息重新投递到原始的 MessageQueue 的时间如果发生消息积压或者 RocketMQ 客户端发生流量管控客户端拉取到消息后进行处理的时间可能会超出预设的延时时间。2026年最新版百万八股面试宝典《Java》《大模型》《大并发》《大数据》本题库由资深讲师联合大厂面试官精心打造专为高级IT岗位设计面试突击体系相关Java企业架构体系相关AI大模型相关HR 面试软技能设计模式相关并发编程相关网络IO与Netty相关互联网三高项目相关亿级流量多级缓存相关数据结构算法相关分布式相关分布式锁相关分布式ID相关核心源码相关大厂线上故障分析相关Docker相关Dubbo相关ElasticSearch相关GoLang相关Java基础核心JVM核心相关JVM调优底层相关Kafka相关Kubernetes相关Linux相关MongoDB相关MQ相关MySQL相关MyBatis相关MyBatisPlus相关Postgresql相关RabbitMQ相关RocketMQ相关Redis相关ShardingSphere相关Shiro相关SkyWalking相关Solr相关Spring 相关SpringBoot相关SpringCloud相关SpringMVC源码相关SpringSecurity相关Spring WebFlux相关领域驱动设计DDD相关任务调度Airflow相关Zookeeper相关HDFS面试题MapReduce面试题Yarn面试题Hive相关SQL面试题相关Scala面试题Spark面试题Flink相关Doris相关Tomcat源码相关网络安全相关运维/云原生相关 
长河君 · 2026/10/23 3:03:14 · 阅读 1.2k 5 张图带你理解 RocketMQ 延时消息机制
延时消息是指发送到 RocketMQ 后不会马上被消费者拉取到而是等待固定的时间才能被消费者拉取到。延时消息的使用场景很多比如电商场景下关闭超时未支付的订单某些场景下需要在固定时间后发送提示消息。1 生产者首先看一个生产者发送延时消息的官方示例代码public static void main(String[] args) throws Exception { // Instantiate a producer to send scheduled messages DefaultMQProducer producer new DefaultMQProducer(ExampleProducerGroup); // Launch producer producer.start(); int totalMessagesToSend 100; for (int i 0; i totalMessagesToSend; i) { Message message new Message(TestTopic, (Hello scheduled message i).getBytes()); // This message will be delivered to consumer 10 seconds later. message.setDelayTimeLevel(3); // Send the message producer.send(message); } // Shutdown producer after use. producer.shutdown(); }从上面的代码可以看到跟普通消息不一样的是消息设置 setDelayTimeLevel 属性值这里设置为 3这里最终将 3 这个延时级别复制给了 DELAY 属性。关于延时级别可以看下面这个定义//MessageStoreConfig类 private String messageDelayLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h;这里延时级别有 18 个上面的示例代码中延迟级别是 3消息会延迟 10s 后消费者才能拉取。2 Broker 处理2.1 写入消息Broker 收到消息后会将消息写入 CommitLog。在写入时会判断消息 DELAY 属性是否大于 0。代码如下//CommitLog 类 if (msg.getDelayTimeLevel() 0) { if (msg.getDelayTimeLevel() this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) { msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()); } topic TopicValidator.RMQ_SYS_SCHEDULE_TOPIC; int queueId ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel()); // Backup real topic, queueId 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())); msg.setTopic(topic); msg.setQueueId(queueId); }从上面的代码可以看到CommitLog 写入时并没有直接写入而是把 Topic 改为 SCHEDULE_TOPIC_XXXX把 queueId 改为延时级别减 1。因为延时级别有 18 个所以这里有 18 个队列。如下图2.2 调度消息延时消息写入后会有一个调度任务不停地拉取这些延时消息这个逻辑在类 ScheduleMessageService。这个类的初始化代码如下public void start() { if (started.compareAndSet(false, true)) { this.load(); this.deliverExecutorService new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl(ScheduleMessageTimerThread_)); //省略部分逻辑 for (Map.EntryInteger, Long entry : this.delayLevelTable.entrySet()) { Integer level entry.getKey(); Long timeDelay entry.getValue(); Long offset this.offsetTable.get(level); if (null offset) { offset 0L; } if (timeDelay ! null) { //省略部分逻辑 this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS); } } //省略持久化的逻辑 } }上面的 load() 方法会加载一个 delayLevelTableConcurrentHashMap类型key 保存延时级别从 1 开始value 保存延时时间(单位是 ms)。load() 方法结束后创建了一个有 18 个核心线程的定时线程池然后遍历 delayLevelTable创建 18 个任务DeliverDelayedMessageTimerTask进行每个延时级别的任务调度。任务调度的代码逻辑如下public void executeOnTimeup() { ConsumeQueue cq ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel)); if (cq null) { this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_WHILE); return; } SelectMappedBufferResult bufferCQ cq.getIndexBuffer(this.offset); if (bufferCQ null) { //省略部分逻辑 this.scheduleNextTimerTask(resetOffset, DELAY_FOR_A_WHILE); return; } long nextOffset this.offset; try { int i 0; ConsumeQueueExt.CqExtUnit cqExtUnit new ConsumeQueueExt.CqExtUnit(); for (; i bufferCQ.getSize() isStarted(); i ConsumeQueue.CQ_STORE_UNIT_SIZE) { long offsetPy bufferCQ.getByteBuffer().getLong(); int sizePy bufferCQ.getByteBuffer().getInt(); long tagsCode bufferCQ.getByteBuffer().getLong(); //省略部分逻辑 long now System.currentTimeMillis(); long deliverTimestamp this.correctDeliverTimestamp(now, tagsCode); nextOffset offset (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); long countdown deliverTimestamp - now; if (countdown 0) { this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); return; } MessageExt msgExt ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy); if (msgExt null) { continue; } MessageExtBrokerInner msgInner ScheduleMessageService.this.messageTimeup(msgExt); //事务消息判断省略 boolean deliverSuc; //只保留同步 deliverSuc this.syncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy); if (!deliverSuc) { this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); return; } } nextOffset this.offset (i / ConsumeQueue.CQ_STORE_UNIT_SIZE); } catch (Exception e) { log.error(ScheduleMessageService, messageTimeup execute error, offset {}, nextOffset, e); } finally { bufferCQ.release(); } this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE); }这段代码可以参考下面的流程图来进行理解上面有一个修正投递时间的函数这个函数的意义是如果已经过了投递时间那么立即投递。代码如下private long correctDeliverTimestamp(final long now, final long deliverTimestamp) { long result deliverTimestamp; long maxTimestamp now ScheduleMessageService.this.delayLevelTable.get(this.delayLevel); if (deliverTimestamp maxTimestamp) { result now; } return result; }注意消息从 CommitLog 转发到 ConsumeQueue 时会判断是否是延时消息Topic SCHEDULE_TOPIC_XXXX 并且延时级别大于 0如果是延时消息就会修改 tagsCode 值为消息投递的时间戳而 tagsCode 原值是 tag 的 HashCode。代码如下//CommitLog类checkMessageAndReturnSize方法 if (delayLevel 0) { tagsCode this.defaultMessageStore.getScheduleMessageService().computeDeliverTimestamp(delayLevel, storeTimestamp); }如下图而 ScheduleMessageService 调度线程将消息从 ConsumeQueue 重新投递到原始队列中时会把 tagsCode 再次修改为 tag 的 HashCode代码如下//类MessageExtBrokerInner这个方法被 messageTimeup 方法调用。 public static long tagsString2tagsCode(final TopicFilterType filter, final String tags) { if (null tags || tags.length() 0) { return 0; } return tags.hashCode(); }如下图2.3 一个问题如果有一个业务场景要求延时消息 3 小时才能消费而 RocketMQ 的延时消息最大延时级别只支持延时 2 小时怎么处理这里提供两个思路供大家参考在 Broker 上修改 messageDelayLevel 的默认配置在客户端缓存 msgId先设置延时级别是 182h当客户端拉取到消息后首先判断有没有缓存如果有缓存则再次发送延时消息这次延时级别是 171h如果没有缓存则进行消费。3 总结经过上面的讲解延时消息的处理流程如下最后延时消息的延时时间并不精确这个时间是 Broker 调度线程把消息重新投递到原始的 MessageQueue 的时间如果发生消息积压或者 RocketMQ 客户端发生流量管控客户端拉取到消息后进行处理的时间可能会超出预设的延时时间。2026年最新版百万八股面试宝典《Java》《大模型》《大并发》《大数据》本题库由资深讲师联合大厂面试官精心打造专为高级IT岗位设计面试突击体系相关Java企业架构体系相关AI大模型相关HR 面试软技能设计模式相关并发编程相关网络IO与Netty相关互联网三高项目相关亿级流量多级缓存相关数据结构算法相关分布式相关分布式锁相关分布式ID相关核心源码相关大厂线上故障分析相关Docker相关Dubbo相关ElasticSearch相关GoLang相关Java基础核心JVM核心相关JVM调优底层相关Kafka相关Kubernetes相关Linux相关MongoDB相关MQ相关MySQL相关MyBatis相关MyBatisPlus相关Postgresql相关RabbitMQ相关RocketMQ相关Redis相关ShardingSphere相关Shiro相关SkyWalking相关Solr相关Spring 相关SpringBoot相关SpringCloud相关SpringMVC源码相关SpringSecurity相关Spring WebFlux相关领域驱动设计DDD相关任务调度Airflow相关Zookeeper相关HDFS面试题MapReduce面试题Yarn面试题Hive相关SQL面试题相关Scala面试题Spark面试题Flink相关Doris相关Tomcat源码相关网络安全相关运维/云原生相关