15、RocketMQ延迟消息

淩亂°似流年 2024-02-05 17:19 19阅读 0赞

概念

延迟消息是指发到MQ的消息不会马上被消费者拉取到，而等待时间才会被消费者拉取到，相比于RabbitMQ与Kafka这个延迟消息是RocketMQ独有的，Rabbitmq只能通过死信队列或插件来实现延迟的效果。

msg.setDelayTimeLevel(3)
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

注意点：4.x版本使用预设的延迟时间是不太灵活的，所以到了5.x版中预设到具体时间戳，按秒的精度来进行定时发送。这固定的18个延迟级别虽然⽆法调整，但是每个延迟级别对应的延迟时间其实是可以调整的。只需要修改截图中的参数就⾏。不过通常不建议这么做。

延迟消息的难点其实是性能，需要不断进⾏定时轮训，全部扫描所有消息是不可能的，而RocketMQ的实现⽅式是预设⼀个系统Topic，名字叫做SCHEDULE\_TOPIC\_XXXX。在这个Topic下，预设18个延迟队列。然后每次只针对这18个队列⾥的消息进⾏延迟操作，不需要⼀直扫描所有的消息。

**源码分析**

在BrokerController构建的时候会去通过this.messageStore.load()加载磁盘文件，其中包括commit log、Consumer Queue、Index File等文件，将它们都加载到内存中，在初始化DefaultMessageStore方法是会去创建ScheduleMessageService类，在load方法里面就会去执行ScheduleMessageService类的load方法，加载延迟消息数据。

初始化delayLevelTable和offsetTable。

public boolean load() {
        
        // 执行父类方法，将多个消息文件加载到内存的offsetTable中
        boolean result = super.load();
        // 解析延迟级别加入delayLevelTable
        result = result && this.parseDelayLevel();
        // 矫正偏移量
        result = result && this.correctDelayOffset();
        return result;
    }

加载延迟delayOffset文件

// 加载消息文件
    public boolean load() {
        
        String fileName = null;
        try {
        
            fileName = this.configFilePath();
            String jsonString = MixAll.file2String(fileName);
    
            if (null == jsonString || jsonString.length() == 0) {
        
                return this.loadBak();
            } else {
        
                this.decode(jsonString);
                log.info("load " + fileName + " OK");
                return true;
            }
        } catch (Exception e) {
        
            log.error("load " + fileName + " failed, and try to load backup file", e);
            return this.loadBak();
        }
    }
    // 获取消息文件名
    public String configFilePath() {
        
        return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig()
            .getStorePathRootDir());
    }
    // 返回消息文件名
    public static String getDelayOffsetStorePath(final String rootDir) {
        
        return rootDir + File.separator + "config" + File.separator + "delayOffset.json";
    }
    
    //  json字符串进行反序列化然后存入offsetTable内存中
    public void decode(String jsonString) {
        
        if (jsonString != null) {
        
            DelayOffsetSerializeWrapper delayOffsetSerializeWrapper =
                DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);
            if (delayOffsetSerializeWrapper != null) {
        
                this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());
            }
        }
    }

解析延迟等级

public boolean parseDelayLevel() {
        
        HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
        // 设置时间单位 秒，分，时，天
        timeUnitTable.put("s", 1000L);
        timeUnitTable.put("m", 1000L * 60);
        timeUnitTable.put("h", 1000L * 60 * 60);
        timeUnitTable.put("d", 1000L * 60 * 60 * 24);
        // 获取内置的messageDelayLevel
        // private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
        try {
        
            // 将对应等级与时间存入delayLevelTable中
            String[] levelArray = levelString.split(" ");
            for (int i = 0; i < levelArray.length; i++) {
        
                String value = levelArray[i];
                String ch = value.substring(value.length() - 1);
                Long tu = timeUnitTable.get(ch);
    
                int level = i + 1;
                if (level > this.maxDelayLevel) {
        
                    this.maxDelayLevel = level;
                }
                long num = Long.parseLong(value.substring(0, value.length() - 1));
                long delayTimeMillis = tu * num;
                this.delayLevelTable.put(level, delayTimeMillis);
                if (this.enableAsyncDeliver) {
        
                    this.deliverPendingTable.put(level, new LinkedBlockingQueue<>());
                }
            }
        } catch (Exception e) {
        
            log.error("parseDelayLevel exception", e);
            log.info("levelString String = {}", levelString);
            return false;
        }
    
        return true;
    }

在Broker的start方法中获取调用DefaultMessageStore类的start方法，在执行时就会去启动延迟消息的服务即执行handleScheduleMessageService方法，默认只能Master开启。

//延迟消息转储服务，当系统消息到事件就会将消息刷入进来存入的topic
    public void start() {
        
        // CAS保证服务只能启动一次
        if (started.compareAndSet(false, true)) {
        
            // 重新加载一次
            this.load();
            // 初始化线程池默认18个线程
            this.deliverExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
            if (this.enableAsyncDeliver) {
        
                this.handleExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
            }
            //扫描延迟队列
            for (Map.Entry<Integer, 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) {
        
                    if (this.enableAsyncDeliver) {
        
                        this.handleExecutorService.schedule(new HandlePutResultTask(level), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
                    }
                    //消息转入原始的queue，启动定时任务，执行run方法
                    this.deliverExecutorService.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME, TimeUnit.MILLISECONDS);
                }
            }
          // 构建一个延迟队列消费偏移量持久化的定时调度任务，首次延迟1000ms之后执行，
            // 后续每次执行间隔flushDelayOffsetInterval时间，默认10s
            this.deliverExecutorService.scheduleAtFixedRate(new Runnable() {
        
                @Override
                public void run() {
        
                    try {
        
                        if (started.get()) {
        
                            ScheduleMessageService.this.persist();
                        }
                    } catch (Throwable e) {
        
                        log.error("scheduleAtFixedRate flush exception", e);
                    }
                }
            }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval(), TimeUnit.MILLISECONDS);
        }
    }

该方法主要是初始化执行的线程池默认18个，以及构建DeliverDelayedMessageTimerTask调度任务，同时构建一个延迟队列消费偏移量持久化的定时调度任务。  
DeliverDelayedMessageTimerTask创建延迟消息任务：当消息被发送到Broker后，根据消息的延迟时间，DeliverDelayedMessageTimerTask就会去执行run方法将消息存储在合适的延迟队列中，并根据设置的延迟时间等待延迟时间过去将消息从延迟队列中取出，并将其重写发送到消费者队列中，以便于消费者可以收到消息并进行消费。  
存入延迟队列的时机：  
在Broker端写入CommitLog文件时执行asyncPutMessage方法（这个前面有过分析）

// 延迟消息转到系统Topic
    if (msg.getDelayTimeLevel() > 0) {
        
        if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
        
            msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
        }
        // 默认系统topic
        topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
        int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    
        // Backup real topic, queueId
        // 移除原有Topic加入新的系统Topic
        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);
    }
    
    // 后台线程写入延迟消息
    DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);

当发现消息设置了延迟级别，首先传入的设置级别计算出系统的延迟级别，然后设置新的系统默认的topic和queueId，最后在doReput方法里面的checkMessageAndReturnSize方法将消息的索引信息写入ConsumerQueue文件中（而存放tagCode的在延时时间未结束之前都表示该消息的延迟结束时间）。  
定时将消息存入与取出executeOnTimeup方法

public void executeOnTimeup() {
        
        // 找到延迟队列对应的ConsumeQueue文件
        // 查询延迟基本delayLevel2QueueId
        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;
        }
        //通过计算，找到这一次扫描需要处理的ConsumeQueue文件
        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();
            //循环过滤ConsumeQueue文件当中的每一条消息索引
            for (; i < bufferCQ.getSize() && isStarted(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
        
                //解析每一条ConsumeQueue记录
                long offsetPy = bufferCQ.getByteBuffer().getLong();
                int sizePy = bufferCQ.getByteBuffer().getInt();
                long tagsCode = bufferCQ.getByteBuffer().getLong();
    
                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);
                    }
                }
                // 计算延迟时间
                // 如果这个消息还未到达延迟结束时间，那么deliverTimestamp就是当前时间
          // 反之如果这个消息已经到达了延迟结束时间，那么deliverTimestamp就是延迟结束时间
                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);
                // 不支持事务消息
                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(), nextOffset, offsetPy, sizePy);
                } else {
        
                    deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy);
                }
                //转储失败就等一等。
                if (!deliverSuc) {
        
                    this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
                    return;
                }
            }
            //计算下一次扫描时的offSet起点。
            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);
    }

查询延迟队列对应的ConsumerQueue文件如果没有找到终止本次任务更新offset，延迟100ms进行下次任务。  
通过计算找到这次扫描需要处理的ConsumerQueue文件，进行遍历解析出每一条ConsumerQueue记录判断延迟时间是否到期，通过延迟结束时间与当前时间作对比，如果延迟结束时间大于当前时间，则表示该条消息还不能被消费，否则延迟结束，可以被消费。  
消息到达延迟结束时间，那么就把消息的原始的topic写入CommitLog文件中，然后更新offsetTable中这个延迟队列的消费进度，消息未达到延迟时间，重新提交TimeTask，延迟执行时间为设置的延迟执行时间为消息剩余的延时时间。