RocketMQ延迟消息实现的方式以及具体实践

骑猪看日落 2022-09-09 10:58 445阅读 0赞

场景：如果一个商城业务需要实现以下功能，用户在下单后，如果用户未付款的情况下，30分钟后需要取消订单，让你来设计你会怎么设计？

首先，针对这个问题实现的方式比较多，往小了说，如果是单机服务，不考虑持久化什么的，完全可以在服务中自主实现：

1.使用JDK自带的 DelayQueue

其实说是用DelayQueue，只是基于了这个queue的排序的特点，如果用 PaiorityQueue ，也就是常说的优先队列一样可以做到。

以下是DelayQueue的例子代码，可以做到延迟消费，并且有随机性。

OrderMessage 消息类：

import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author: Urey
     * @Date: 2021/8/31
     */
    // 所有的消息必须实现Delayed类，具体可以看DelayQueue的源码，泛型有限制
    public class OrderMessage implements Delayed {
    
        private long id;
        private long createdAt;
        private long expiredAt;
        private int delayLevel;
    
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.toMillis(getExpiredAt() - System.currentTimeMillis());
        }
    
        public long getId() {
            return id;
        }
    
        public void setId(long id) {
            this.id = id;
        }
    
        public long getCreatedAt() {
            return createdAt;
        }
    
        public void setCreatedAt(long createdAt) {
            this.createdAt = createdAt;
        }
    
        public long getExpiredAt() {
            return expiredAt;
        }
    
        public void setExpiredAt(long expiredAt) {
            this.expiredAt = expiredAt;
        }
    
        @Override
        public int compareTo(Delayed o) {
            return Long.compare(this.getExpiredAt(),((OrderMessage)o).getExpiredAt());
        }
    
        public OrderMessage(long id,long expiredAt,int delayLevel,long createdAt) {
            this.id = id;
            this.expiredAt = expiredAt;
            this.delayLevel = delayLevel;
            this.createdAt = createdAt;
        }
    
        @Override
        public String toString() {
            return "OrderMessage{" +
                    "id=" + id +
                    ", createdAt=" + createdAt +
                    ", expiredAt=" + expiredAt +
                    ", delayLevel=" + delayLevel +
                    '}';
        }
    }

具体代码实现类：

import java.util.Random;
    import java.util.concurrent.DelayQueue;
    
    /**
     * @Author: Urey
     * @Date: 2021/8/31
     */
    public class DelayService {
    
        static DelayQueue<OrderMessage> delayQueue = new DelayQueue();
    
        static long duration = 1000;
    
        public static void main(String[] args) throws InterruptedException {
            Thread thread = new Thread(() -> {
                for (int i = 0;i <= 10;i++){
                    long current = System.currentTimeMillis();
                    int delayLevel = (new Random().nextInt(10) + 1);
                    delayQueue.offer(new OrderMessage(i,current + delayLevel * duration,delayLevel,current));
                    System.out.println("add task success!current is " + current + " and delayLevel is " + delayLevel);
                }
            });
            thread.start();
            Thread thread2 = new Thread(() -> {
                for (;;) {
                    if (delayQueue.isEmpty()){
                        try {
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
    
                    }else {
                        if (delayQueue.peek().getExpiredAt() <= System.currentTimeMillis()) {
                            OrderMessage message = delayQueue.poll();
                            System.out.println("message is " + message.toString() + " and time is pass " + (System.currentTimeMillis() - message.getCreatedAt()));
                        }
                    }
                }
            });
            thread2.start();
            Thread.sleep(1);
            while (true){
                if (delayQueue.isEmpty()) {
                    System.out.println("任务完成！退出任务！");
                    break;
                }
            }
            System.exit(0);
        }
    }

以下为本地console输出的日志：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16][]

可以看到，任务只要往DelayQueue里面添加，thread2线程会一直循环去拿第一个任务，直到任务满足要求会消费当前任务。

基于Java来直接实现延迟队列操作简单，代码易懂，实现起来也不是很难。但是缺点在于如果队列消息过多，会导致OOM的存在，另外就是提到的持久化问题，导致消息丢失，集群更不用说了，因为是基于Java内存操作，所以集群基本不可能实现，除非借助Mysql或者Redis这种。

2.使用Redis的回调

Redis我们都知道，是可以做缓存的作用，还有缓存的过期等等。

Redis在2.8版本以后提供了回调，意思就是当我们的字符串过期了，那么Redis会主动通知我们。

比如我们可以在Redis里面设置多个Key，这些Key对应了我们的待消费的消费ID，并且设置延迟消费的过期时间，当Redis的Key失效了，那么此时Redis会主动通知服务，当我们的服务接收到Redis的通知以后可以做出回应，以达到延迟队列的目的。

借用别人的博客看一下Redis具体的回调实现

# [springboot整合redis缓存过期回调函数实现][springboot_redis] #

使用Redis的好处在于可以实现持久化，集群方式下也可以做到延迟的问题。缺点我能想到的大概是依赖于其他组件，每次都需要回调，会造成很多无效的请求以及如果数据量过大，会占用许多的Redis的Key(这部分可能总结的不是很好，完全是个人理解)。

Redis另外也有自身的延迟队列实现，这个具体的可以自行百度下，因为文章想主要叙说MQ的实现。

3.基于RocketMQ实现

RocketMQ和RabbitMQ都可以实现延迟队列，本文只基于RocketMQ来叙说，毕竟RocketMQ是基于Java开发的，我本身也是Java程序猿。

首先在Rocket-Console中，我们可以看到有许多的Topic，这些Topic有些是RocketMQ默认在启动时会自行创建，有些是我们自己所创建的。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 1][]上图中的SCHEDULE\_TOPIC\_XXXX就是我们今天的主角了。

SCHEDULE\_TOPIC\_XXXX顾名思义就是一个有定时任务的Topic，这个是RocketMQ默认创建的Topic。

我们在设置DelayLevel以后，RocketMQ会默认帮我们发送延迟消息，延迟消息和我们的producer以及consumer都是没有关系的，是broker的commitLog在搞鬼，我们可以根据具体的源码了解一下。

下面是producer端的代码

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_15_color_FFFFFF_t_70_g_se_x_16][]

可以看到，producer在发送完消息后，仍然是发送给了BLOG\_TOPIC这个我们指定的topic。因为我们指定的delayLevel是4，所以默认是30秒后消费。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

可以看到，我们的consumer端确实是在30秒后才消费到了消息。

以下来剖析下具体的原理来分析下RocketMQ是怎样做到的。

Producer在发送消息时，指定的Topic只是意在表明，我这条消息是需要发往RocketMQ的BLOG\_TOPIC的，但是Broker在接收到消息后会将消息进行持久化处理，也就是要走到CommitLog.class类，当是DelayMessage时，CommitLog会执行以下逻辑

if (msg.getDelayTimeLevel() > 0) {
                    if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                        msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                    }
    
                    topic = ScheduleMessageService.SCHEDULE_TOPIC;
                    queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    
                    // Backup real 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()));
    
                    msg.setTopic(topic);
                    msg.setQueueId(queueId);
                }

跟踪Debug我们可以看到，CommitLog将消息再一次封装了，原先的MQ的消息都放到了properties的变量中，我们原先需要发送的topic现在对应的是REAL\_TOPIC，而当前消息要发送的topic是SCHEDULE\_TOPIC\_XXXX。也就是说Broker此时会先将消息发送至SCHEDULE\_TOPIC\_XXXX来做一个暂存。  
![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_18_color_FFFFFF_t_70_g_se_x_16][]

消息放置完成以后是需要消费的，在启动Broker的时候， ScheduleMessageService.class的类会加载默认所有的delay级别的线程，同时会不断的创建，销毁，创建，销毁线程的方式去循环调用，而我们常说的RocketMQ不支持自主的设置delayMessage的时间正是因为这个，每个delayLevel会创建一个线程，也就是RocketMQ默认有18个等级，那么需要创建18个线程循环调用

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 4][]

下图即是delayLevel对应的延迟时间，如delayLevel是1 对应的即是延迟1000毫秒消费

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_16_color_FFFFFF_t_70_g_se_x_16][]

随机截取的部分线程循环的日志

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 5][]

当修改Broker的messageDelayLevel = ”6h“配置以后，可以看到当前就只会有一个线程running

当消息触发以后，该线程会将消息取出，调用putMessage方法，将该消息再发送至对应的topic，至此就完成了延迟消费的实现。

MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
    PutMessageResult putMessageResult = ScheduleMessageService.this.defaultMessageStore
                                                    .putMessage(msgInner);

以上就是针对RoketMQ的延迟消息的分析实现，所以借助于这种思想，我们在自定义实现的时候其实也很简单，就是基于一个排序的队列，然后一个线程去循环获取head的数据，满足要求消费即可。

如有错误，欢迎指出，及时更正。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/abab693aa97e41a0bee2071230304b66.png
[springboot_redis]: https://blog.csdn.net/weixin_43944391/article/details/89476343
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/e75b54fc49cf4e968476d7db333761d0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/a541590d756f4be596cd595e81edca1a.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/ff6ff9c373584725811af07a26ed6107.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/1e2c70bb80424173ba240f75d2aceb49.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_18_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/1c37b1ed4f2c45578a00a84d5e0e3d4e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/38f51ee286d949829f5925db408a1dec.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_16_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/967a3b53f83f4c4fb1bbef1cac399e5d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBATGVvbmlz5Li2TA_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220829/468ea5d761de4045a896d618c55dad1b.png