TDMQ RocketMQ 版秒级定时消息原理解析

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

// 延时消息服务启动public void start() {        if (started.compareAndSet(false, true)) {            this.load();            // 创建与延时级别一样数量的线程            this.deliverExecutorService = ThreadUtils.newScheduledThreadPool(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));            // 初始化线程池等操作...            for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {                // 为每个延时等级都创建定时任务                if (timeDelay != null) {                    if (this.enableAsyncDeliver) {                        this.handleExecutorService.schedule(new HandlePutResultTask(level),                              FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);                    }                    this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level,                         offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);                }            }            // 位点以及数据持久化定时任务...        }    }

• 当延时消息进入时会判断延时等级属性，放进相应的队列尾部。

• 通过线程池对每个队列进行轮询扫描，判断头部消息是否达到延时时间，如果达到则将消息投递到 Real Topic，否则继续轮询。

注意：多级别延时消息受到消息存储时间限制；

// 执行循环扫描public void executeOnTimeUp() {        ConsumeQueueInterface cq;        // 获取cq，如果cq为null则新建定时任务并校准下次扫描开始位点        ....        // 设置本次扫描开始位点        long nextOffset = this.offset;        try {             while (bufferCQ.hasNext() && isStarted()) {                    CqUnit cqUnit = bufferCQ.next();                    long offsetPy = cqUnit.getPos();                    int sizePy = cqUnit.getSize();                    long tagsCode = cqUnit.getTagsCode();                    // 判断是否达到延时时间                    long now = System.currentTimeMillis();                    long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);                    // 计算下次拉取offset                    long currOffset = cqUnit.getQueueOffset();                    nextOffset = currOffset + cqUnit.getBatchNum();                    long countdown = deliverTimestamp - now;                    if (countdown > 0) {                        // 未到延时时间，则创建下次定时任务并return                        this.scheduleNextTimerTask(currOffset, DELAY_FOR_A_WHILE);                        ScheduleMessageService.this.updateOffset(this.delayLevel, currOffset);                        return;                    }                    MessageExt msgExt = ScheduleMessageService.this.brokerController.                        getMessageStore().lookMessageByOffset(offsetPy, sizePy);                    // 获取消息内容并重新投递                    if (ScheduleMessageService.this.enableAsyncDeliver) {                        this.asyncDeliver(msgInner, msgExt.getMsgId(), currOffset, offsetPy, sizePy);                    } else {                        this.syncDeliver(msgInner, msgExt.getMsgId(), currOffset, offsetPy, sizePy);                    }                    //如果投递失败，立即创建下次任务，并设置下次任务开始扫描位点为currOffset，防止消息丢失                 }            } catch (Exception e) {                ......            } finally {                bufferCQ.release();            }            this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);        }

// 在Producer端设置消息为延时消息Message msg = new Message();msg.setTopic("TopicA");msg.setTags("Tag");msg.setBody("this is a delay message".getBytes());// 设置延迟level为5，对应延迟1分钟msg.setDelayTimeLevel(5);producer.send(msg);

/** * Represents a slot of timing wheel. Format: * ┌────────────┬───────────┬───────────┬───────────┬───────────┐ * │delayed time│ first pos │ last pos  │    num    │   magic   │ * ├────────────┼───────────┼───────────┼───────────┼───────────┤ * │   8bytes   │   8bytes  │  8bytes   │   4bytes  │   4bytes  │ * └────────────┴───────────┴───────────┴───────────┴───────────┘ */

• TimerLog 是定时消息索引文件，保存定时消息的索引（在消息文件中存储的位置），内部通过反向链表进行链接，它的写入为 Append-only Log，保证了消息写入的性能。

public final static int UNIT_SIZE = 4  //size            + 8 //prev pos            + 4 //magic value            + 8 //curr write time, for trace            + 4 //delayed time, for check            + 8 //offsetPy            + 4 //sizePy            + 4 //hash code of real topic            + 8; //reserved value, just in case of

public boolean enqueue(int queueId) {        // 判断服务是否正常执行,获取cq文件准备遍历        ConsumeQueueInterface cq = this.messageStore.getConsumeQueue(TIMER_TOPIC, queueId);        // 校准位点...        long offset = currQueueOffset;        ReferredIterator<CqUnit> iterator = null;        try {            iterator = cq.iterateFrom(offset);            if (null == iterator) {                return false;            }            // 开始遍历索引文件并从CommitLog读取出消息体            int i = 0;            while (iterator.hasNext()) {                i++;                try {                    CqUnit cqUnit = iterator.next();                    // 计算消息读取位点与消息大小                    MessageExt msgExt = getMessageByCommitOffset(offsetPy, sizePy);                    if (null == msgExt) {                        perfCounterTicks.getCounter("enqueue_get_miss");                    } else {                        // 读取消息的相关信息进行timerRequest构造并投递到enqueuePutQueue                        TimerRequest timerRequest = new TimerRequest(offsetPy, sizePy,                            delayedTime, System.currentTimeMillis(), MAGIC_DEFAULT, msgExt);                        while (!enqueuePutQueue.offer(timerRequest, 3, TimeUnit.SECONDS)) {                            if (!isRunningEnqueue()) {                                return false;                            }                        }                     }                     // 指标计算等逻辑...                    }                } catch (Exception e) {...} finally {...}                // if broker role changes, ignore last enqueue                if (!isRunningEnqueue()) {                    return false;                }                currQueueOffset = offset + i;            }            currQueueOffset = offset + i;            return i > 0;        } catch (Exception e) {...} finally {            if (iterator != null) {                iterator.release();            }        }        return false;    }

b.  另一个线程 TimerEnqueuePutService 将其执行 Timerlog-unit 构建逻辑并放入 TimerLog，更新时间轮（Timewheel）的存储内容。

    protected void putMessageToTimerWheel(TimerRequest req) {            try {                // 指标计算...+如果消息已经到期直接投递到dequeuePutQueue等待投递到realTopic                                if (shouldRunningDequeue && req.getDelayTime() < currWriteTimeMs) {                    req.setEnqueueTime(Long.MAX_VALUE);                    dequeuePutQueue.put(req);                } else {                    boolean doEnqueueRes = doEnqueue(                        req.getOffsetPy(), req.getSizePy(), req.getDelayTime(), req.getMsg());                    req.idempotentRelease(doEnqueueRes || storeConfig.isTimerSkipUnknownError());                }             } catch (Throwable t) {...}    }    public boolean doEnqueue(long offsetPy, int sizePy, long delayedTime, MessageExt messageExt) {        // 调整deleayTime等操作...        String realTopic = messageExt.getProperty(MessageConst.PROPERTY_REAL_TOPIC);        Slot slot = timerWheel.getSlot(delayedTime);        // 构造timeLog unit        ByteBuffer tmpBuffer = timerLogBuffer;        tmpBuffer.clear();        tmpBuffer.putInt(TimerLog.UNIT_SIZE); //size        //...        long ret = timerLog.append(tmpBuffer.array(), 0, TimerLog.UNIT_SIZE);        if (-1 != ret) {            // 如果timelog更新成功则更新timewheel            timerWheel.putSlot(delayedTime, slot.firstPos == -1 ? ret : slot.firstPos, ret,                isDelete ? slot.num - 1 : slot.num + 1, slot.magic);        }        return -1 != ret;    }

2.  针对取出定时消息的 Service，每50ms读取下一秒的 Slot。有三个线程将读取到的消息重新放回 CommitLog。

a.  首先，TimerDequeueGetService 每50ms读一次下一秒的 Slot，从 TimerLog 中得到指定的数据，并放进 dequeueGetQueue。

public int dequeue() throws Exception {        // 判断消息是否正常出队        Slot slot = timerWheel.getSlot(currReadTimeMs);        try {            long currOffsetPy = slot.lastPos;            Set<String> deleteUniqKeys = new ConcurrentSkipListSet<>();            LinkedList<TimerRequest> normalMsgStack = new LinkedList<>();            LinkedList<TimerRequest> deleteMsgStack = new LinkedList<>();            LinkedList<SelectMappedBufferResult> sbrs = new LinkedList<>();            SelectMappedBufferResult timeSbr = null;            //read the timer log one by one， 开始遍历当前时间的所有消息            while (currOffsetPy != -1) {                // 读取出timelog对应的unit                if (null == timeSbr || timeSbr.getStartOffset() > currOffsetPy) {                    timeSbr = timerLog.getWholeBuffer(currOffsetPy);                    if (null != timeSbr) {                        sbrs.add(timeSbr);                    }                }                if (null == timeSbr) {                    break;                }                long prevPos = -1;                try {                    // 读取buffer获取相关数据并构造timerRequest                    TimerRequest timerRequest = new TimerRequest(offsetPy, sizePy,                        delayedTime, enqueueTime, magic);                    timerRequest.setDeleteList(deleteUniqKeys);                    // 判断消息是否不用滚动并是删除类型，是就加进deleteMsgStack表示删除类型消息                    if (needDelete(magic) && !needRoll(magic)) {                        deleteMsgStack.add(timerRequest);                    } else {                        normalMsgStack.addFirst(timerRequest);                    }                } catch (Exception e) {...} finally {//计算下次读取timelog位点，以及指标计算}            }            // 下面先投递delete类型msg,确保可以精准删除对应消息            CountDownLatch deleteLatch = new CountDownLatch(deleteMsgStack.size());            //read the delete msg: the msg used to mark another msg is deleted            for (List<TimerRequest> deleteList : splitIntoLists(deleteMsgStack)) {                for (TimerRequest tr : deleteList) {                    tr.setLatch(deleteLatch);                }                dequeueGetQueue.put(deleteList);            }            //do we need to use loop with tryAcquire            checkDequeueLatch(deleteLatch, currReadTimeMs);            CountDownLatch normalLatch = new CountDownLatch(normalMsgStack.size());            //read the normal msg            for (List<TimerRequest> normalList : splitIntoLists(normalMsgStack)) {                for (TimerRequest tr : normalList) {                    tr.setLatch(normalLatch);                }                dequeueGetQueue.put(normalList);            }            checkDequeueLatch(normalLatch, currReadTimeMs);            } catch (Throwable t) {...}        return 1;    }

Message message = new Message(TOPIC, ("Hello" + i).getBytes(StandardCharsets.UTF_8));// 延迟 10s 后投递message.setDelayTimeSec(10);// 延迟 10000ms 后投递，投递到服务端后计算定时时间，即投递到服务端的时间+delayTimemessage.setDelayTimeMs(10_000L);// 定时投递，定时时间为当前时间 + 10000msmessage.setDeliverTimeMs(System.currentTimeMillis() + 10_000L);// 发送消息SendResult result = producer.send(message);