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.Entry 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() 方法会加载一个 delayLevelTable（ConcurrentHashMap类型），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 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();}

如下图：