RocketMQ源码 发送 延迟消息 源码分析

前言

rocketMQ 支持的延迟消息，简单理解就是对于生产者发送的消息，支持设置固定时间的延迟级别，在到达指定的延迟时间时，才会投递到消费者队列，消费者才能消费到消息。

延迟队列和普通消息的发送流程，主要流程都是一致的，区别在于：

可以参考源码架构图来看，在?DledgerCommitLog 组件写入消息之前，会针对设置了setDelayTimeLevel 延迟级别的消息，改写? topic 为?RMQ_SYS_SCHEDULE_TOPIC，那么?ReputMessageService 消息重分发现场就会将消息投递到?RMQ_SYS_SCHEDULE_TOPIC 下。

再由 ScheduleMessageService 调度线程，定时将?RMQ_SYS_SCHEDULE_TOPIC 下的消息取出，判断是否到达延迟时间，如果到达延迟时间，就将消息重新投递到真实的Topic中，这样消费者就能消费到真是的消息。

源码版本：4.9.3

源码架构图

源码分析

1. 源码入口

本地源码入口从一个代码example工程中的一个示例，开始org.apache.rocketmq.example.schedule.ScheduledMessageProducer#main

public class ScheduledMessageProducer {
    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();
    }
    
}

... 发送流程和普通消息一致，对通用流程感兴趣可以看下另一篇文章，RocketMQ源码 发送消息源码分析-CSDN博客...

2. DledgerCommitLog 存储消息

broker接受的消息后，会调用?DLedgerCommitLog 存储组件，进行消息存储写入 CommitLog组件。在写的过程中有一个 setMessageInfo 信息的函数，会改写 topic 为?RMQ_SYS_SCHEDULE_TOPIC。

org.apache.rocketmq.store.dledger.DLedgerCommitLog#asyncPutMessage

    // 异步写入消息
    @Override
    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {

        // 存储统计服务
        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

        // 事务类型
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());

        // 设置消息信息：包含延迟消息的处理，改写延迟消息topic和queueId
        setMessageInfo(msg, tranType);

        final String finalTopic = msg.getTopic();
    }

    private void setMessageInfo(MessageExtBrokerInner msg, int tranType) {
        // Set the storage time
        msg.setStoreTimestamp(System.currentTimeMillis());
        // Set the message body BODY CRC (consider the most appropriate setting
        // on the client)
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));

        //should be consistent with the old version
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {

            // 延时消息，延迟投递
            // Delay Delivery
            // 如果延时级别大于0
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }

                // 系统级延时调度topic，用于延时调度投递
                String topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
                // 根据延时级别获取对应的队列,队列的id为delayLevel-1
                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()));

                // 设置topic和queueId
                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }

        InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
        if (bornSocketAddress.getAddress() instanceof Inet6Address) {
            msg.setBornHostV6Flag();
        }

        InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
        if (storeSocketAddress.getAddress() instanceof Inet6Address) {
            msg.setStoreHostAddressV6Flag();
        }
    }

...?ReputMessageService 重分发服务会定时将延迟消息投递到?RMQ_SYS_SCHEDULE_TOPIC ...

3. ScheduleMessageService 定时调度消息服务

投递延迟消息定时任务，会定时 100ms 扫描?RMQ_SYS_SCHEDULE_TOPIC 主题下的队列中的消息，计算消息的延迟时间，是不是已经到期，如果到期，就将消息投递倒真实的 Topic中。

org.apache.rocketmq.store.schedule.ScheduleMessageService.DeliverDelayedMessageTimerTask

    // 投递延时消息定时任务
    class DeliverDelayedMessageTimerTask implements Runnable {
        // 延时级别
        private final int delayLevel;
        // 偏移量
        private final long offset;

        public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {
            this.delayLevel = delayLevel;
            this.offset = offset;
        }

        @Override
        public void run() {
            try {
                if (isStarted()) {
                    // 执行延时消息投递
                    this.executeOnTimeup();
                }
            } catch (Exception e) {
                // XXX: warn and notify me
                log.error("ScheduleMessageService, executeOnTimeup exception", e);
                this.scheduleNextTimerTask(this.offset, DELAY_FOR_A_PERIOD);
            }
        }

        /**
         * 计算正确的延时投递时间，如果延时消息已经过期，返回当前时间
         * @return
         */
        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;
        }

        // 执行延时消息投递
        public void executeOnTimeup() {
            // 根据topic、queueId获取对应的CQ
            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;
            }

            // 根据偏移量读取到CQ内存映射数据
            SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
            if (bufferCQ == null) {
                long resetOffset;
                if ((resetOffset = cq.getMinOffsetInQueue()) > this.offset) {
                    log.error("schedule CQ offset invalid. offset={}, cqMinOffset={}, queueId={}",
                        this.offset, resetOffset, cq.getQueueId());
                } else if ((resetOffset = cq.getMaxOffsetInQueue()) < this.offset) {
                    log.error("schedule CQ offset invalid. offset={}, cqMaxOffset={}, queueId={}",
                        this.offset, resetOffset, cq.getQueueId());
                } else {
                    resetOffset = this.offset;
                }

                this.scheduleNextTimerTask(resetOffset, DELAY_FOR_A_WHILE);
                return;
            }

            long nextOffset = this.offset;
            try {
                int i = 0;
                ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();

                // 遍历CQ内存映射队列
                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(); // deliverTimestamp

                    if (cq.isExtAddr(tagsCode)) {
                        if (cq.getExt(tagsCode, cqExtUnit)) {
                            tagsCode = cqExtUnit.getTagsCode();
                        } else {
                            //can't find ext content.So re compute tags code.
                            log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
                                tagsCode, offsetPy, sizePy);
                            long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
                            tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
                        }
                    }

                    long now = System.currentTimeMillis();
                    // 计算正确的延时投递时间
                    long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);

                    // 下一个消息的偏移量 = 起始偏移量 + 遍历的步长
                    nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);

                    long countdown = deliverTimestamp - now;
                    // 如果延时时间还未到，则继续等待，100毫秒后再次执行
                    if (countdown > 0) {
                        this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
                        return;
                    }

                    // 读取消息
                    MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
                    if (msgExt == null) {
                        continue;
                    }

                    // 转换消息，还原消息真实的topic
                    MessageExtBrokerInner msgInner = ScheduleMessageService.this.messageTimeup(msgExt);
                    if (TopicValidator.RMQ_SYS_TRANS_HALF_TOPIC.equals(msgInner.getTopic())) {
                        log.error("[BUG] the real topic of schedule msg is {}, discard the msg. msg={}",
                            msgInner.getTopic(), msgInner);
                        continue;
                    }

                    boolean deliverSuc;
                    if (ScheduleMessageService.this.enableAsyncDeliver) {
                        deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
                    } else {
                        // 将消息同步投递到，真实的topic中
                        deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), offset, offsetPy, sizePy);
                    }

                    if (!deliverSuc) {
                        // 如果投递失败，则继续等待，100毫秒后再次执行
                        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);
        }

        // 调度下一个延时投递消息
        public void scheduleNextTimerTask(long offset, long delay) {
            // 延时投递消息
            ScheduleMessageService.this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(
                this.delayLevel, offset), delay, TimeUnit.MILLISECONDS);
        }

        // 同步投递消息
        private boolean syncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
            int sizePy) {
            PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, false);
            PutMessageResult result = resultProcess.get();
            boolean sendStatus = result != null && result.getPutMessageStatus() == PutMessageStatus.PUT_OK;
            if (sendStatus) {
                ScheduleMessageService.this.updateOffset(this.delayLevel, resultProcess.getNextOffset());
            }
            return sendStatus;
        }

        private boolean asyncDeliver(MessageExtBrokerInner msgInner, String msgId, long offset, long offsetPy,
            int sizePy) {
            Queue<PutResultProcess> processesQueue = ScheduleMessageService.this.deliverPendingTable.get(this.delayLevel);

            //Flow Control
            int currentPendingNum = processesQueue.size();
            int maxPendingLimit = ScheduleMessageService.this.defaultMessageStore.getMessageStoreConfig()
                .getScheduleAsyncDeliverMaxPendingLimit();
            if (currentPendingNum > maxPendingLimit) {
                log.warn("Asynchronous deliver triggers flow control, " +
                    "currentPendingNum={}, maxPendingLimit={}", currentPendingNum, maxPendingLimit);
                return false;
            }

            //Blocked
            PutResultProcess firstProcess = processesQueue.peek();
            if (firstProcess != null && firstProcess.need2Blocked()) {
                log.warn("Asynchronous deliver block. info={}", firstProcess.toString());
                return false;
            }

            PutResultProcess resultProcess = deliverMessage(msgInner, msgId, offset, offsetPy, sizePy, true);
            processesQueue.add(resultProcess);
            return true;
        }

        // 异步投递消息
        private PutResultProcess deliverMessage(MessageExtBrokerInner msgInner, String msgId, long offset,
            long offsetPy, int sizePy, boolean autoResend) {
            CompletableFuture<PutMessageResult> future =
                ScheduleMessageService.this.writeMessageStore.asyncPutMessage(msgInner);
            return new PutResultProcess()
                .setTopic(msgInner.getTopic())
                .setDelayLevel(this.delayLevel)
                .setOffset(offset)
                .setPhysicOffset(offsetPy)
                .setPhysicSize(sizePy)
                .setMsgId(msgId)
                .setAutoResend(autoResend)
                .setFuture(future)
                .thenProcess();
        }
    }