RockerMQ4.x 延时消息原理、流程梳理~

叁歲伎倆 2023-10-13 04:46 6阅读 0赞

# 前言 #

相信用过RocketMq的小伙伴都知道，RocketMq支持延时消息，通过设置指定的延时级别就可以让消息实现不同时效的延时功能，今天带大家了解的就是延时消息的原理~

--------------------

# 如何实现延时？ #

在带大家正式了解RocketMq延时原理之前，先问大家一个问题，如果我们自己来实现延时功能，我们会如何实现？

1.  **Sleep**

**经典永不过时**，相信Sleep是我们最早接触的具有延时功能的函数，下面代码就可以简单实现延时5s后执行业务逻辑

java复制代码public static void main(String[] args) throws InterruptedException {
       TimeUnit.SECONDS.sleep(5);
    
       System.out.println("执行业务代码");
    }

1.  **Timer**

Timer 类是在 JDK 1.3 版本中引入的。它位于 java.util 包中，用于支持简单的定时任务调度。

不过Timer也有着许多缺陷，谨慎使用~

java复制代码public static void main(String[] args) throws InterruptedException {
    
      // schedule实现延时
      Timer timer = new Timer();
      timer.schedule(new TimerTask() {
        @Override
        public void run() {
          System.out.println("执行业务逻辑");
        }
      }, 5000);
    
    }

1.  **ScheduledThreadPoolExecutor**

ScheduledThreadPoolExecutor 类是在 JDK 1.5 版本中引入的。它是 ThreadPoolExecutor 类的子类，专门用于支持定时任务的调度和执行。

java复制代码public static void main(String[] args) throws InterruptedException {
    
      ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
    
      // schedule实现延时
      executor.schedule(() -> {
        System.out.println("执行业务代码");
      }, 5, TimeUnit.SECONDS);
    
    }

1.  **时间轮**

有关时间轮我就不细说了，在各大开源框架，诸如： Netty、Dubbo、Kafka都少不了它的影子

--------------------

# RocketMq如何实现的延时消息？ #

在上面带大家了解了常见的延时方案之后，我们再来探索RcketMq的延时原理~

--------------------

# RocketMq官方延时案例 #

下面是RocketMq官方提供的案例，我们可以看到代码中通过setDelayTimeLevel设定了延时级别，对应的延时时间就是10s

java复制代码package org.apache.rocketmq.example.schedule;
    
    public class ScheduledMessageProducer {
    
      public static final String PRODUCER_GROUP = "ExampleProducerGroup";
      public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
      public static final String TOPIC = "TestTopic";
    
      public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
    
        producer.start();
        int totalMessagesToSend = 100;
        for (int i = 0; i < totalMessagesToSend; i++) {
          Message message = new Message(TOPIC, ("Hello scheduled message " + i).getBytes());
    
          // 设定延时级别
          message.setDelayTimeLevel(3);
    
          // 发送消息
          SendResult result = producer.send(message);
          System.out.print(result);
        }
    
        // Shutdown producer after use.
        producer.shutdown();
      }
    
    }

![format_png][]

--------------------

# 存储延时消息 #

Producer回将消息发送到Broker，接下来看看Broker是怎么处理延时消息的.

org.apache.rocketmq.store.CommitLog\#asyncPutMessage中会对延时消息进行特殊处理

1.  如果不是事务消息且延时级别 > 0，说明该消息是延时消息，需要进行特殊处理
2.  检查延时级别是否超过最大值18，如果超过则重置为18
3.  保存消息的原始topic和queueId，并将该消息的topic覆盖为延时消息专用的topic，且将queueId设置为延时级别 - 1

java复制代码public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
     
      // ......
    
      String topic = msg.getTopic();
      
      final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
      // 事务消息不支持延时
      if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
          || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        
        // 如果延时级别 > 0说明是延时消息
        if (msg.getDelayTimeLevel() > 0) {
          if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
            msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
          }
    
          // 延时消息使用同一个topic: SCHEDULE_TOPIC_XXXX
          // 且queueId: 延时级别 - 1
          topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
          int queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    
          // 存储原始消息的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);
        }
      }
      
      // ......
    }

经过上面的一番桶特殊处理及之后一些逻辑后，消息会被存储到commitLog中去

--------------------

# 取出延时消息 #

Broker将延迟消息写到了commitLog中后，由于Broker替换了消息原始topic，所以订阅该topic的消费者此时还无法消费该消息。

org.apache.rocketmq.store.schedule.ScheduleMessageService，Broker启动时会启动ScheduleMessageService

而在ScheduleMessageService中会为每个延时级别都开启一个延时任务，延时能力正是利用我前面提到的ScheduledThreadPoolExecutor

同时也会开启一个定时任务，固定时间持久化每个队列的消费偏移量

java复制代码public class ScheduleMessageService extends ConfigManager {
      
      private static final long FIRST_DELAY_TIME = 1000L;
    
      // Broker启动时会初始化这个Map,key是延迟等级，共计18个
      // key: 延时级别，value: 延时时间（毫秒）
      private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =
            new ConcurrentHashMap<Integer, Long>(32);
      
      // key: 延时级别,value: 消费偏移量
      private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
            new ConcurrentHashMap<Integer, Long>(32);
    
      public void start() {
        if (started.compareAndSet(false, true)) {
          this.load();
          this.deliverExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageTimerThread_"));
          if (this.enableAsyncDeliver) {
            this.handleExecutorService = new ScheduledThreadPoolExecutor(this.maxDelayLevel, new ThreadFactoryImpl("ScheduleMessageExecutorHandleThread_"));
          }
          
          // 遍历延时级别map
          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;
            }
    
            // 为每个延时级别开一个延时任务，延时1s执行，使用的是ScheduledThreadPoolExecutor
            if (timeDelay != null) {
              // enableAsyncDeliver 默认false
              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);
            }
          }
    
          // 消费偏移量持久化定时任务,每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);
        }
      }
    }

DeliverDelayedMessageTimerTask封装了delayLevel、offset，实现了Runnable重写了run方法，在run方法中如果处于运行中，则调用executeOnTimeup方法执行具体逻辑

java复制代码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);
        }
      }
    }

java复制代码public void executeOnTimeup() {
      
      // 根据topic和延时级别找到对应的ConsumeQueue
      ConsumeQueue cq =
        ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
                                                                         delayLevel2QueueId(delayLevel));
      // ......
    
      long nextOffset = this.offset;
      try {
        int i = 0;
        ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
        for (; i < bufferCQ.getSize() && isStarted(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
          
          // ......
    
          // 当前时间
          long now = System.currentTimeMillis();
          // 计算投递时间，时间存储在了tag hashcode 中
          long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
          nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
    
          // 剩余时间
          long countdown = deliverTimestamp - now;
          if (countdown > 0) {
            // 还未到投递时间，重新加入到延时任务，此时延时100ms
            this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
            return;
          }
    
          // 从commitLog读取message
          MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);
          if (msgExt == null) {
            continue;
          }
    
          // 恢复message原始的topic和queueId，之前保存过
          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;
          }
    
          // 将消息再次写入commitlog中，topic是原始topic,这样消费者就可以去消费了
          boolean deliverSuc;
          // enableAsyncDeliver默认为false
          if (ScheduleMessageService.this.enableAsyncDeliver) {
            deliverSuc = this.asyncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy);
          } else {
            deliverSuc = this.syncDeliver(msgInner, msgExt.getMsgId(), nextOffset, offsetPy, sizePy);
          }
    
          // 写入commitLog失败，延时重试
          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();
      }
    
      // 更新offset，重新添加延时任务处理延时消息
      this.scheduleNextTimerTask(nextOffset, DELAY_FOR_A_WHILE);
    }

1.  根据topic和延时级别找到对应的ConsumeQueue
2.  计算校验消息是否到达投递时间
    
    1.  如果还没到达则重新加入延时任务
    2.  到达投递时间，则从commitLog中取出消息，将消息的topic、queueId设置为原始值，并重新写入到commitLog中
3.  更新消费偏移量nextOffset，并重新添加延时任务处理延时消息

--------------------

# 总结 #

RocketMq延时消息大致流程并不复杂，下面简单总结下

1.  当我们需要发送延时消息的时候，需要对消息设置延时级别，标明该消息是延时消息
2.  Broker收到延时消息后，会重新设置延时消息的topic、queueId，并备份消息原始的topic、queueId，之后写入到commitLog
3.  Broker会启动ScheduleMessageService，ScheduleMessageService会为每个延时级别启动一个延时任务(利用ScheduledThreadPoolExecutor)
    
    1.  根据topic、延时级别找到对应的consumerQueue，然后检查消息是否到达投递时间
        
        1.  如果还没达到投递时间，则重新添加到延时任务中
        2.  如果已经达到投递时间，则将消息topic、queueId归还为原始值，再重新写入commitLog，这样消费者就能感知到进行消费了。
4.  最后更新消费偏移量，并重新加入到延时任务进行下一次处理

[format_png]: https://img-blog.csdnimg.cn/img_convert/2a83adc070606773e52713c0b612f1c0.webp?x-oss-process=image/format,png