RocketMQ原理学习--延时消息实现原理

阳光穿透心脏的1/2处 2022-04-05 17:26 388阅读 0赞 
  1.      RocketMQ提供了延时消息类型，简单来说就是生产者在发送消息的时候指定一个延时时间，当到达延时时间之后消息才能够被投送到消费者。
  3.   首先我们可以考虑一下RocketMQ的延时消息是如何实现：

（1）生产者将延时消息发送到Broker，Broker是如何区分普通消息和延时消息（消息类型）

（2）消息如何保证不被消费者拉取到的（RocketMQ将消息以SCHEDULE_TOPIC_XXXX为topic将延时消息持久化，等到达延时消息之后再以原有的topic重新保存）。

1、简单示例

（1）RocketMQ目前指定的延时时间间隔有1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h，用等级来表示时间间隔。

  1. public class DelayProducer {
  3.     public static void main(String[] args) throws MQClientException, InterruptedException {
  4.         DefaultMQProducer producer = new DefaultMQProducer("rmq-group");
  5.         producer.setNamesrvAddr("localhost:9876");
  6.         producer.start();
  7.         try {
  8.             for (int i = 0; i < 3; i++) {
  9.                 Message msg = new Message("TopicA-test",// topic
  10.                         "TagA",// tag
  11.                         (new Date() + "Hello RocketMQ ,QuickStart 11" + i)
  12.                                 .getBytes()// body
  13.                 );
  14.                 //1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h。
  15.                 // level=0，表示不延时。level=1，表示 1 级延时，对应延时 1s。level=2 表示 2 级延时，对应5s，以此类推
  16.                 msg.setDelayTimeLevel(2);
  18.                 SendResult sendResult = producer.send(msg);
  19.                 System.out.println(sendResult);
  20.             }
  21.         } catch (Exception e) {
  22.             e.printStackTrace();
  23.         }
  24.        producer.shutdown();
  25.     }
  27. }

2、Broker持久化延时消息

  1.      Broker对于接收到的消息首先会判断一下是不是延时消息，如果是延时消息会将消息以SCHEDULE\_TOPIC\_XXXX为topic替换原有的topic名称进行持久化，实现方法在CommitLog的putMessage中。
  3.     首先会判断msg的延时标准如果大于0，则重新设置消息的topic名称和queueId，之后将消息以SCHEDULE\_TOPIC\_XXXX为topic，以延时时间的等级为queueId持久化到commitlog文件中。
  5. // Delay Delivery
  6.             if (msg.getDelayTimeLevel() > 0) {
  7.                 if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
  8.                     msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
  9.                 }
  11.                 topic = ScheduleMessageService.SCHEDULE_TOPIC;
  12.                 queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
  14.                 // Backup real topic, queueId
  15.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
  16.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
  17.                 msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
  19.                 msg.setTopic(topic);
  20.                 msg.setQueueId(queueId);
  21.             }

3、延时消息处理

  1.     Broker将延时消息以SCHEDULE\_TOPIC\_XXXX为topic名称将消息进行持久化，接下来我们看看Broker是如何将消息在延时消息到达之后进行消息还原的。
  3.     RocketMQ提供了定时任务服务ScheduleMessageService，通过定时任务的方式不断的读取topic为SCHEDULE\_TOPIC\_XXXX何queueId为延时等级的消息进行消息还原处理，这样消息被还原之后消费者就可以拉取消息了。

每个消费等级有个定时任务DeliverDelayedMessageTimerTask：

  1. for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
  2.             Integer level = entry.getKey();
  3.             Long timeDelay = entry.getValue();
  4.             Long offset = this.offsetTable.get(level);
  5.             if (null == offset) {
  6.                 offset = 0L;
  7.             }
  9.             if (timeDelay != null) {
  10.                 this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
  11.             }
  12.         }

在DeliverDelayedMessageTimerTask中根据SCHEDULE_TOPIC_XXXX名称和延时等级对应的queueId获取消息队列，然后从commitlog中读取消息，还原消息的原有信息（消息的原topic信息）再将消息持久化到commitlog文件中，这样消费者就可以拉取消息了。

  1. public void executeOnTimeup() {
  2.             //获取消费者消息
  3.             ConsumeQueue cq =
  4.                 ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,
  5.                     delayLevel2QueueId(delayLevel));
  7.             long failScheduleOffset = offset;
  9.             if (cq != null) {
  10.                 //读取消息
  11.                 SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
  12.                 if (bufferCQ != null) {
  13.                     try {
  14.                         long nextOffset = offset;
  15.                         int i = 0;
  16.                         ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
  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);
  29.                                     long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
  30.                                     tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
  31.                                 }
  32.                             }
  34.                             long now = System.currentTimeMillis();
  35.                             long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);
  37.                             nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
  39.                             long countdown = deliverTimestamp - now;
  41.                             if (countdown <= 0) {
  42.                                 //获取消息
  43.                                 MessageExt msgExt =
  44.                                     ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
  45.                                         offsetPy, sizePy);
  47.                                 if (msgExt != null) {
  48.                                     try {
  49.                                         //还原消息信息topic名称等
  50.                                         MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
  51.                                         //重新将消息持久化到commitlog中
  52.                                         PutMessageResult putMessageResult =
  53.                                             ScheduleMessageService.this.defaultMessageStore
  54.                                                 .putMessage(msgInner);
  56.                                         if (putMessageResult != null
  57.                                             && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
  58.                                             continue;
  59.                                         } else {
  60.                                             // XXX: warn and notify me
  61.                                             log.error(
  62.                                                 "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
  63.                                                 msgExt.getTopic(), msgExt.getMsgId());
  64.                                             ScheduleMessageService.this.timer.schedule(
  65.                                                 new DeliverDelayedMessageTimerTask(this.delayLevel,
  66.                                                     nextOffset), DELAY_FOR_A_PERIOD);
  67.                                             ScheduleMessageService.this.updateOffset(this.delayLevel,
  68.                                                 nextOffset);
  69.                                             return;
  70.                                         }
  71.                                     } catch (Exception e) {
  72.                                         /*
  73.                                          * XXX: warn and notify me
  77.                                          */
  78.                                         log.error(
  79.                                             "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
  80.                                                 + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
  81.                                                 + offsetPy + ",sizePy=" + sizePy, e);
  82.                                     }
  83.                                 }
  84.                             } else {
  85.                                 ScheduleMessageService.this.timer.schedule(
  86.                                     new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
  87.                                     countdown);
  88.                                 ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
  89.                                 return;
  90.                             }
  91.                         } // end of for
  93.                         nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
  94.                         ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
  95.                             this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
  96.                         ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
  97.                         return;
  98.                     } finally {
  100.                         bufferCQ.release();
  101.                     }
  102.                 } // end of if (bufferCQ != null)
  103.                 else {
  105.                     long cqMinOffset = cq.getMinOffsetInQueue();
  106.                     if (offset < cqMinOffset) {
  107.                         failScheduleOffset = cqMinOffset;
  108.                         log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
  109.                             + cqMinOffset + ", queueId=" + cq.getQueueId());
  110.                     }
  111.                 }
  112.             } // end of if (cq != null)
  114.             ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
  115.                 failScheduleOffset), DELAY_FOR_A_WHILE);
  116.         }

总结：

  1.      延时消息的实现还是挺精巧的，首先将延时消息换了一个topic名称进行持久化，这样消费者就无法获取消息，然后有定时任务，会将消息还原到原有的topic信息，这样消费者又可以重新拉取消息了。

发表评论

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

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

相关阅读

    相关 RocketMQ进阶-延时消息

        前言 在开发中经常会遇到延时任务的需求，例如在12306购买车票，若生成订单30分钟未支付则自动取消；还有在线商城完成订单后48小时不评价 ，自动5星好评。像

    水深无声水深无声/ 0 赞/ 164 阅读

    相关 RocketMQ原理学习--消息类型

    一、消费模式 集群消费：当使用集群消费模式时，MQ 认为任意一条消息只需要被集群内的任意一个消费者处理即可。 广播消费：当使用广播消费模式时，MQ 会将每条消息推送给

    末蓝、末蓝、/ 0 赞/ 402 阅读