RocketMQ源码解析之延迟消息实现原理

Love The Way You Lie 2022-12-27 07:48 251阅读 0赞

原创不易，转载请注明出处

文章目录

      • 前言
      • 1.延时消息的demo
      • 2.实现的原理

前言

今天要谈论的话题其实非常轻松，但是我们有些业务场景是离不开它的，其实说到延迟消息，不知道大家有没有想到它的业务场景，我这里说几个，比如我们购物下单，然后多长时间没有付款就会取消，或者我们下单打车，多长时间没有调度到车订单取消，不知道大家怎样实现这些业务场景，比如说我一个线程，然后一直扫订单表，扫出过期的订单啥的，但是这种方式会有一个问题，那就是订单量大的时候会给数据库带来很大的压力，而且很low，这种场景我们完全可以交给RocketMQ的延迟消息来处理。本文将从RocketMQ的延迟消息demo讲起，然后解释下其实现原理。

1.延时消息的demo

这里我们直接写一个RocketMQ消息生产者发送延时消息的一个demo

  1. public class ScheduledMessageProducer { 
  4.     public static void main(String[] args) throws Exception { 
  5.         // 实例化一个生产者来产生延时消息
  6.         DefaultMQProducer producer = new DefaultMQProducer("ExampleProducerGroup");
  7.         // 启动生产者
  8.         producer.start();
  9.         int totalMessagesToSend = 100;
  10.         for (int i = 0; i < totalMessagesToSend; i++) { 
  11.             Message message = new Message("TestTopic", ("Hello scheduled message " + i).getBytes());
  12.             // 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel)
  13.             message.setDelayTimeLevel(3);
  14.             // 发送消息
  15.             producer.send(message);
  16.         }
  17.         // 关闭生产者
  18.         producer.shutdown();
  19.     }
  20. }

其实这个demo非常简单，与发送普通消息区别不大，唯一区别的就是需要你在Message里面设置延迟等级，默认的延迟等级与时间对应关系如下

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

也就是延迟等级1级对应的是1s，2级是5s，以此类推。如果你想改变这个等级的话，你可以设置broker的messageDelayLevel参数，s代表秒，m是分钟，h是小时，这里有点要注意的是，你消息消费失败重试也是要走延迟消息的，这块你需要注意下，然后别改变里延迟消息规则，然后造成失败重试有问题，默认失败重试规则就是3+失败重试次数的延时等级

2.实现的原理

实现原理其实挺简单的，在消息生产者端，你设置这延迟消息等级，其实就是往消息的property设置一个DELAY的key与value值，key就是DELAY，然后value 就是你延迟的等级，这就是消息生产者端要做的事情，接下来介绍下broker端的实现原理。
在broker 端，收到一个消息，在往commitlog进行putMessage的时候，也就是追加写入消息的时候，会检查这个消息的延迟等级，也就是下面这端代码
在这里插入图片描述
会将你的topic设置成SCHEDULE_TOPIC_XXXX，然后queueId设置成你延时等级-1 ，其实你有没有发现，他的实现与事务消息的差不多，都是修改topic 与queueId。
同时，在broker启动的时候，它会启动一个定时任务，定义一个timer，设置几个定时任务（定时任务的数量与延迟等级是有关系的，一个任务对应一个定时任务）不断的扫SCHEDULE_TOPIC_XXXX这个topic的消息。
在这里插入图片描述
接着我们看下这scheduleMessageService 的start方法的实现
在这里插入图片描述
其实这个start方法就是遍历这延时等级表，然后往timer里面添加延时任务，最后创建一个定时任务，每10s执行一次持久化任务。这个持久化任务其实不用担心，就是往磁盘里面写下，延时等级对应的offset。
接着我们看下DeliverDelayedMessageTimerTask这个任务的 实现
在这里插入图片描述
可以看到run方法里面执行了executeOnTimeup这个方法

  1. /// 取出consumeQueue
  2.             ConsumeQueue cq =
  3.                 ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,
  4.                         delayLevel2QueueId(delayLevel));
  5.             long failScheduleOffset = offset;
  6.             if (cq != null) { 
  8.                 /// 取出
  9.                 SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
  10.                 if (bufferCQ != null) { 
  11.                     try { 
  12.                         long nextOffset = offset;
  13.                         int i = 0;
  14.                         ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
  16.                         ///遍历取
  17.                         for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) { 
  18.                             long offsetPy = bufferCQ.getByteBuffer().getLong();
  19.                             int sizePy = bufferCQ.getByteBuffer().getInt();
  20.                             long tagsCode = bufferCQ.getByteBuffer().getLong();
  22.                             if (cq.isExtAddr(tagsCode)) { 
  23.                                 if (cq.getExt(tagsCode, cqExtUnit)) { 
  24.                                     tagsCode = cqExtUnit.getTagsCode();
  25.                                 } else { 
  26.                                     //can't find ext content.So re compute tags code.
  27.                                     log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
  28.                                         tagsCode, offsetPy, sizePy);
  31.                                     // 消息存入时间
  32.                                     long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
  34.                                     // 计算交付时间
  35.                                     tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
  36.                                 }
  37.                             }
  38.                             long now = System.currentTimeMillis();
  39.                             // 计算正确的交付时间戳
  40.                             long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
  41.                             nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
  42.                             long countdown = deliverTimestamp - now;
  44.                             if (countdown <= 0) { 
  46.                                 // 拿出消息
  47.                                 MessageExt msgExt =
  48.                                     ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
  49.                                         offsetPy, sizePy);
  51.                                 if (msgExt != null) { 
  52.                                     try { 
  53.                                         ///转成MessageExtBrokerInner
  54.                                         MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
  56.                                         /// 写到 commitlog中
  57.                                         PutMessageResult putMessageResult =
  58.                                             ScheduleMessageService.this.writeMessageStore
  59.                                                 .putMessage(msgInner);
  61.                                         if (putMessageResult != null
  62.                                             && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { 
  63.                                             continue;
  64.                                         } else { 
  65.                                             // XXX: warn and notify me 失败
  66.                                             log.error(
  67.                                                 "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
  68.                                                 msgExt.getTopic(), msgExt.getMsgId());
  69.                                             // 这么做主要是为了防止os cache 繁忙
  70.                                             ScheduleMessageService.this.timer.schedule(
  71.                                                 new DeliverDelayedMessageTimerTask(this.delayLevel,
  73.                                                     /// 延迟一段时间 10000L = 10s
  74.                                                     nextOffset), DELAY_FOR_A_PERIOD);
  76.                                             /// 更新 这个等级的offset
  77.                                             ScheduleMessageService.this.updateOffset(this.delayLevel,
  78.                                                 nextOffset);
  79.                                             return;
  80.                                         }
  81.                                     } catch (Exception e) { 
  82.                                         /* * XXX: warn and notify me */
  83.                                         log.error(
  84.                                             "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
  85.                                                 + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
  86.                                                 + offsetPy + ",sizePy=" + sizePy, e);
  87.                                     }
  88.                                 }
  89.                             } else { 
  91.                                 // 重新添加任务
  92.                                 ScheduleMessageService.this.timer.schedule(
  93.                                     new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
  94.                                     countdown);
  96.                                 // 更新offset
  97.                                 ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
  98.                                 return;
  99.                             }
  100.                         } // end of for
  102.                         nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
  103.                         ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
  104.                                 //100ms
  105.                             this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
  106.                         // 更新
  107.                         ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
  108.                         return;
  109.                     } finally { 
  111.                         bufferCQ.release();
  112.                     }
  113.                 } // end of if (bufferCQ != null)
  114.                 else { 
  115.                     // 找出最小的
  116.                     long cqMinOffset = cq.getMinOffsetInQueue();
  117.                     if (offset < cqMinOffset) { 
  119.                         failScheduleOffset = cqMinOffset;
  120.                         log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
  121.                             + cqMinOffset + ", queueId=" + cq.getQueueId());
  122.                     }
  123.                 }
  124.             } // end of if (cq != null)
  126.             ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
  127.                 failScheduleOffset), DELAY_FOR_A_WHILE);

这个方法首先是获取对应延迟等级的consumeQueue这个队列，取出offset往后的消息，进行遍历，找出每个unit对应的commitlog真实offset，然后通过commitlog offset 从commitlog获取到真实的那个消息，根据它的存储实现与延迟时间，算出与真实交付时间的差值，如果是小于等于0的话，说明延迟时间到了，这个时候就要暴露给消息消费者了，它就会将topic与queueId转成之前的那个topic queueId，然后重新扔到commitlog中，这个时候通过reput线程的dispatch处理，消息消费者就能发现这个消息并消费，如果这个差值还不够的话，重新创建调度任务，然后延迟执行时间是这个差值，扔到timer中。

发表评论

表情：
评论列表 （有 0 条评论，251人围观）

还没有评论，来说两句吧...

相关阅读