public class Consumer {
    public static void main(String[] args) throws MQClientException {
        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
        // 實(shí)例化消費(fèi)者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("OneMoreGroup");
        // 設(shè)置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
        // 訂閱一個(gè)或者多個(gè)Topic，以及Tag來(lái)過(guò)濾需要消費(fèi)的消息
        consumer.subscribe("OneMoreTopic", "*");
        // 注冊(cè)回調(diào)實(shí)現(xiàn)類來(lái)處理從broker拉取回來(lái)的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            System.out.printf("%s %s Receive New Messages:%n"
                    , sdf.format(new Date())
                    , Thread.currentThread().getName());
            for (MessageExt msg : msgs) {
                System.out.printf("\tMsg Id: %s%n", msg.getMsgId());
                System.out.printf("\tBody: %s%n", new String(msg.getBody()));
            }
            // 標(biāo)記該消息已經(jīng)被成功消費(fèi)
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 啟動(dòng)消費(fèi)者實(shí)例
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

public class DelayProducer {
    public static void main(String[] args) throws Exception {
        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss.SSS");
        // 實(shí)例化消息生產(chǎn)者Producer
        DefaultMQProducer producer = new DefaultMQProducer("OneMoreGroup");
        // 設(shè)置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 啟動(dòng)Producer實(shí)例
        producer.start();
        Message msg = new Message("OneMoreTopic"
                , "DelayMessage", "This is a delay message.".getBytes());
        //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        //設(shè)置消息延遲級(jí)別為3，也就是延遲10s。
        msg.setDelayTimeLevel(3);
        // 發(fā)送消息到一個(gè)Broker
        SendResult sendResult = producer.send(msg);
        // 通過(guò)sendResult返回消息是否成功送達(dá)
        System.out.printf("%s Send Status: %s, Msg Id: %s %n"
                , sdf.format(new Date())
                , sendResult.getSendStatus()
                , sendResult.getMsgId());
        // 如果不再發(fā)送消息，關(guān)閉Producer實(shí)例。
        producer.shutdown();
    }
}

10:37:14.992 Send Status: SEND_OK, Msg Id: C0A8006D5AB018B4AAC216E0DB690000

10:37:25.026 ConsumeMessageThread_1 Receive New Messages:
	Msg Id: C0A8006D5AB018B4AAC216E0DB690000
	Body: This is a delay message.

// 延遲級(jí)別大于0，就是延時(shí)消息
if (msg.getDelayTimeLevel() > 0) {
    // 判斷當(dāng)前延遲級(jí)別，如果大于最大延遲級(jí)別，
    // 就設(shè)置當(dāng)前延遲級(jí)別為最大延遲級(jí)別。
    if (msg.getDelayTimeLevel() > this.defaultMessageStore
            .getScheduleMessageService().getMaxDelayLevel()) {
        msg.setDelayTimeLevel(this.defaultMessageStore
                .getScheduleMessageService().getMaxDelayLevel());
    }
    // 獲取延遲消息的主題，
    // 其中RMQ_SYS_SCHEDULE_TOPIC的值為SCHEDULE_TOPIC_XXXX
    topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
    // 根據(jù)延遲級(jí)別獲取延遲消息的隊(duì)列Id，
    // 隊(duì)列Id其實(shí)就是延遲級(jí)別減1
    queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    // 備份真正的主題和隊(duì)列Id
    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()));
    // 設(shè)置延時(shí)消息的主題和隊(duì)列Id
    msg.setTopic(topic);
    msg.setQueueId(queueId);
}

public void start() {
    // 使用AtomicBoolean確保start方法僅有效執(zhí)行一次
    if (started.compareAndSet(false, true)) {
        this.timer = new Timer("ScheduleMessageTimerThread", true);
        // 遍歷所有延遲級(jí)別
        for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
            // key為延遲級(jí)別
            Integer level = entry.getKey();
            // value為延遲級(jí)別對(duì)應(yīng)的毫秒數(shù)
            Long timeDelay = entry.getValue();
            // 根據(jù)延遲級(jí)別獲得對(duì)應(yīng)隊(duì)列的偏移量
            Long offset = this.offsetTable.get(level);
            // 如果偏移量為null，則設(shè)置為0
            if (null == offset) {
                offset = 0L;
            }
            if (timeDelay != null) {
                // 為每個(gè)延遲級(jí)別創(chuàng)建定時(shí)任務(wù)，
                // 第一次啟動(dòng)任務(wù)延遲為FIRST_DELAY_TIME，也就是1秒
                this.timer.schedule(
                        new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
            }
        }
        // 延遲10秒后每隔flushDelayOffsetInterval執(zhí)行一次任務(wù)，
        // 其中，flushDelayOffsetInterval默認(rèn)配置也為10秒
        this.timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    // 持久化每個(gè)隊(duì)列消費(fèi)的偏移量
                    if (started.get()) ScheduleMessageService.this.persist();
                } catch (Throwable e) {
                    log.error("scheduleAtFixedRate flush exception", e);
                }
            }
        }, 10000, this.defaultMessageStore
        	.getMessageStoreConfig().getFlushDelayOffsetInterval());
    }
}

// 根據(jù)主題和隊(duì)列Id獲取消息隊(duì)列
ConsumeQueue cq =
        ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(
                TopicValidator.RMQ_SYS_SCHEDULE_TOPIC
                , delayLevel2QueueId(delayLevel));

// 根據(jù)消費(fèi)偏移量從消息隊(duì)列中獲取所有有效消息
SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);

// 遍歷所有消息
for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
    // 獲取消息的物理偏移量
    long offsetPy = bufferCQ.getByteBuffer().getLong();
    // 獲取消息的物理長(zhǎng)度
    int sizePy = bufferCQ.getByteBuffer().getInt();
    long tagsCode = bufferCQ.getByteBuffer().getLong();
    // 省略部分代碼...
    long now = System.currentTimeMillis();
    // 計(jì)算消息應(yīng)該被消費(fèi)的時(shí)間
    long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
	// 計(jì)算下一條消息的偏移量
    nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE)
	long countdown = deliverTimestamp - now;
    // 省略部分代碼...
}

// 根據(jù)消息的物理偏移量和大小獲取消息
MessageExt msgExt =
    ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
            offsetPy, sizePy);

// 重新構(gòu)建新的消息，包括：
// 1.清除消息的延遲級(jí)別
// 2.恢復(fù)真正的消息主題和隊(duì)列Id
MessageExtBrokerInner msgInner = 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;
}
// 重新把消息發(fā)送到真正的消息隊(duì)列上
PutMessageResult putMessageResult =
        ScheduleMessageService.this.writeMessageStore
                .putMessage(msgInner);