延时消息队列

Myth丶恋晨 2024-03-26 22:39 86阅读 0赞

##  ##

**目录**

前言

一、延时队列实用场景

二、DelayQueue

DelayQueue的实现

使用延迟队列

DelayQueue实现延时任务的优缺点

三、RocketMQ

原理

四、Kafka

原理

实现

DelayMessage定义

消息发送代码

消费者代码

参考

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

## 前言 ##

延时队列的内部是**有序的**，最重要的特性就体现在它的延时属性上，**延时队列就是用来存放需要在指定时间点被处理的元素的队列**。

队列是存储消息的载体，延时队列存储的对象是延时消息。所谓的延时消息，是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费。

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

### **一、**延时队列实用场景 ###

>  *  淘宝七天自动确认收货。在我们签收商品后，物流系统会在七天后延时发送一个消息给支付系统，通知支付系统将货款打给商家，这个过程持续七天，就是使用了消息中间件的延迟推送功能；
>  *  订单在三十分钟之内未支付则自动取消；
>  *  新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒；
>  *  **用户注册成功后，如果三天内没有登陆则进行短信提醒****；**
>  *  用户发起退款，如果三天内没有得到处理则通知相关运营人员；
>  *  预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。

### **二、**DelayQueue ###

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E> {
    }

DelayQueue是一个无界的BlockingQueue，是线程安全的（无界指的是队列的元素数量不存在上限，队列的容量会随着元素数量的增加而扩容，阻塞队列指的是当队列内元素数量为0的时候，试图从队列内获取元素的线程将被阻塞或者抛出异常）

以上是阻塞队列的特点，而延迟队列还拥有自己如下的特点：

DelayQueue中存入的必须是实现了Delayed接口的对象（Delayed定义了一个getDelay的方法，用来判断排序后的元素是否可以从Queue中取出，并且Delayed接口还继承了Comparable用于排序），插入Queue中的数据根据compareTo方法进行排序（DelayQueue的底层存储是一个PriorityQueue，PriorityQueue是一个可排序的Queue，其中的元素必须实现Comparable接口的compareTo方法），并通过getDelay方法返回的时间确定元素是否可以出队，只有小于等于0的元素（即延迟到期的元素）才能够被取出

延迟队列不接收null元素

#### DelayQueue的实现 ####

public class UserDelayTask implements Delayed {
    
        @Getter
        private UserRegisterMessage message;
    
        private long delayTime;
    
        public UserDelayTask(UserRegisterMessage message, long delayTime) {
            this.message = message;
            // 延迟时间加当前时间
            this.delayTime = System.currentTimeMillis() + delayTime;
        }
    
        // 获取任务剩余时间
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        @Override
        public int compareTo(Delayed o) {
            return Long.compare(delayTime, ((UserDelayTask) o).delayTime);
        }
    }

定义延迟队列并交付容器管理

/**
         * 延迟队列
         */
        @Bean("userDelayQueue")
        public DelayQueue<UserDelayTask> orderDelayQueue() {
            return new DelayQueue<UserDelayTask>();
        }

#### 使用延迟队列 ####

@Resource
    private DelayQueue<UserDelayTask> orderDelayQueue;
     
    UserDelayTask task = new UserDelayTask(message, 1000 * 60);
            orderDelayQueue.add(task);

开启线程处理延迟任务

@Override
        public void afterPropertiesSet() throws Exception {
            new Thread(() -> {
                while (true) {
                    try {
                        UserDelayTask task = orderDelayQueue.take();
                        // 当队列为null的时候，poll()方法会直接返回null, 不会抛出异常，但是take()方法会一直等待，
                        // 因此会抛出一个InterruptedException类型的异常。(当阻塞方法收到中断请求的时候就会抛出InterruptedException异常)
                        UserRegisterMessage message = task.getMessage();
                        execute(message);
    
                        // 执行业务
                    } catch (Exception ex) {
                        log.error("afterPropertiesSet", ex);
                    }
                }
            }).start();
        }

#### DelayQueue实现延时任务的优缺点 ####

使用DelayQueue实现延时任务非常简单，而且简便，全部都是标准的JDK代码实现，不用引入第三方依赖（不依赖redis实现、消息队列实现等），非常的轻量级。

它的缺点就是所有的操作都是基于应用内存的，一旦出现应用单点故障，可能会造成延时任务数据的丢失。如果订单并发量非常大，因为DelayQueue是无界的，订单量越大，队列内的对象就越多，可能造成OOM的风险。所以使用DelayQueue实现延时任务，只适用于任务量较小的情况。

### **三、****RocketMQ** ###

RocketMQ 和本身就有延迟队列的功能，但是开源版本只能支持固定延迟时间的消息，不支持任意时间精度的消息（这个好像只有阿里云版本的可以）。

他的默认时间间隔分为 18 个级别，基本上也能满足大部分场景的需要了。

默认延迟级别：1s、 5s、 10s、 30s、 1m、 2m、 3m、 4m、 5m、 6m、 7m、 8m、 9m、 10m、 20m、 30m、 1h、 2h。

使用起来也非常的简单，直接通过`setDelayTimeLevel`设置延迟级别即可。

setDelayTimeLevel(level)

#### **原理** ####

实现原理说起来比较简单，Broker 会根据不同的延迟级别创建出多个不同级别的队列，当我们发送延迟消息的时候，根据不同的延迟级别发送到不同的队列中，同时在 Broker 内部通过一个定时器去轮询这些队列（RocketMQ 会为每个延迟级别分别创建一个定时任务），如果消息达到发送时间，那么就直接把消息发送到指 topic 队列中。

RocketMQ 这种实现方式是放在服务端去做的，同时有个好处就是相同延迟时间的消息是可以保证有序性的。

谈到这里就顺便提一下关于消息消费重试的原理，这个本质上来说其实是一样的，对于消费失败需要重试的消息实际上都会被丢到延迟队列的 topic 里，到期后再转发到真正的 topic 中。

![f5e9fe2ddfc2c2350ebad8a6d5362849.jpeg][]

### 四、**Kafka** ###

对于 Kafka 来说，原生并不支持延迟队列的功能，需要我们手动去实现，这里我根据 RocketMQ 的设计提供一个实现思路。

这个设计，我们也不支持任意时间精度的延迟消息，只支持固定级别的延迟，因为对于大部分延迟消息的场景来说足够使用了。

只创建一个 topic，但是针对该 topic 创建 18 个 partition，每个 partition 对应不同的延迟级别，这样做和 RocketMQ 一样有个好处就是能达到相同延迟时间的消息达到有序性。

#### **原理** ####

*  首先创建一个单独针对延迟队列的 topic，同时创建 18 个 partition 针对不同的延迟级别
 *  发送消息的时候根据消息延迟等级发送到延迟 topic 对应的 partition，同时把原 topic 保存到 延迟消息 中。

*  内嵌的`consumer`单独设置一个`ConsumerGroup`去消费延迟 topic 消息，消费到消息之后如果没有达到延迟时间那么就进行`pause`，然后`seek`到当前`ConsumerRecord`的`offset`位置，同时使用定时器去轮询延迟的`TopicPartition`，达到延迟时间之后进行`resume。`

KafkaConsumer 提供了暂停和恢复的API函数，调用消费者的暂停方法后就无法再拉取到新的消息，同时长时间不消费kafka也不会认为这个消费者已经挂掉了。

*  如果达到了延迟时间，那么就获取到延迟消息中的真实 topic ，直接转发

这里为什么要进行`pause`和`resume`呢？因为如果不这样的话，如果超时未消费达到`max.poll.interval.ms`最大时间（默认300s），那么将会触发 Rebalance。

![be10947db4fbd86230c1da2b8c17712a.jpeg][]

#### 实现 ####

##### DelayMessage定义 #####

/**
     * 延迟消息
     *
     * @author yangyanping
     * @date 2023-08-31
     */
    @Getter
    @Setter
    @ToString
    public class DelayMessage<T> implements DTO {
        /**
         * 消息级别，共18个，对应18个partition
         */
        private Integer level;
    
        /**
         * 业务类型，真实投递到的topic
         */
        private String topic;
    
        /**
         * 目标消息key
         */
        private String key;
    
        /**
         * 事件
         */
        private DomainEvent<T> event;
    }

##### 消息发送代码 #####

public void publishAsync(DelayMessage delayMessage) {
            String topic = "delay_topic";
    
            try {
                Integer level = delayMessage.getLevel();
                Integer delayPartition = level - 1;
                String data = JSON.toJSONString(delayMessage);
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, delayPartition, "", data);
    
                ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);
                future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onSuccess(SendResult<String, String> result) {
                        //发送成功后回调
                        log.info("{}-异步发送成功, result={}。", topic, result.getRecordMetadata().toString());
                    }
    
                    @Override
                    public void onFailure(Throwable throwable) {
                        //发送失败回调
                        log.error("{}-异步发送失败。", topic, throwable);
                    }
                });
            } catch (Exception ex) {
                log.error("{}-异步发送异常。", topic, ex);
            }
        }

##### 消费者代码 #####

import cn.hutool.core.collection.CollectionUtil;
    import cn.hutool.core.convert.Convert;
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.xinwu.shushan.exception.BizException;
    import com.xinwu.shushan.launch.infra.publisher.KafkaPublisher;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.OffsetAndMetadata;
    import org.apache.kafka.common.TopicPartition;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.kafka.listener.ConsumerSeekAware;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.Resource;
    import java.util.*;
    import java.util.concurrent.Executors;
    import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 参考RocketMQ支持延迟消息设计，不支持任意时间精度的延迟消息，只支持特定级别的延迟消息，
     * 将消息延迟等级分为1s、5s、10s 、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h，共18个级别，
     * 只创建一个有18个分区的延时topic，每个分区对应不同延时等级。
     * <p>
     * https://blog.csdn.net/weixin_40270946/article/details/121293032
     * <p>
     * https://zhuanlan.zhihu.com/p/365802989
     *
     * @author yangyanping
     * @date 2023-08-30
     */
    @Slf4j
    @Component
    public class DelayConsumer implements ConsumerSeekAware {
    
        /**
         * 锁
         */
        private final Object lock = new Object();
    
        /**
         * 间隔
         */
        private final int interval = 5000;
    
        private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
    
        /**
         * 消费者
         */
        private volatile Consumer kafkaConsumer;
    
        @Resource
        private KafkaPublisher kafkaPublisher;
    
        @PostConstruct
        public void init() {
            //当系统需要循环间隔一定时间执行某项任务的时候可以使用scheduleWithFixedDelay方法来实现
            executorService.scheduleWithFixedDelay(() -> {
                synchronized (lock) {
                    resume();
                    lock.notifyAll();
                    log.info("DelayConsumer-notifyAll");
                }
            }, 0, interval, TimeUnit.MILLISECONDS);
        }
    
        /**
         * 批量消费消息
         */
        @KafkaListener(topics = "#{'${shushan.launch.event.delayTopic.topic}'}", groupId = "#{'${spring.kafka.consumer.group-id}'}", concurrency = "1")
        public void onMessage(List<ConsumerRecord<String, String>> records, Consumer consumer) {
            synchronized (lock) {
                try {
                    /**
                     * 7、ConcurrentModificationException
                     * java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
                     * 报错内容：线程不安全
                     *
                     * 原因分析：Kafka consumer是非线程安全的
                     */
                    this.kafkaConsumer = consumer;
    
                    if (CollectionUtil.isEmpty(records)) {
                        log.info("DelayConsumer-records is empty !");
                        consumer.commitSync();
                        return;
                    }
    
                    boolean delay = false;
    
                    for (ConsumerRecord<String, String> record : records) {
                        long timestamp = record.timestamp();
                        String value = record.value();
                        JSONObject jsonObject = JSON.parseObject(value);
                        Integer level = Convert.toInt(jsonObject.get("level"));
                        String targetTopic = Convert.toStr(jsonObject.get("topic"));
                        String event = Convert.toStr(jsonObject.get("event"));
                        String msgKey = Convert.toStr(jsonObject.get("key"));
    
                        TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());
                        long delayTime = getDelayTime(timestamp, level);
    
                        if (delayTime <= System.currentTimeMillis()) {
                            log.info("DelayConsumer-delayTime={} <= currentTime={},msgKey={}", delayTime, System.currentTimeMillis(), msgKey);
                            // 处理消息
                            processMessage(record, consumer, topicPartition, targetTopic, msgKey, event);
                        } else {
                            log.info("DelayConsumer-delayTime={} > currentTime={},msgKey={},partition={}", delayTime, System.currentTimeMillis(), msgKey, topicPartition.partition());
                            // 暂停消费
                            consumer.pause(Collections.singletonList(topicPartition));
                            consumer.seek(topicPartition, record.offset());
                            delay = true;
                            break;
                        }
                    }
    
                    if (delay) {
                        lock.wait();
                    }
                } catch (Exception var10) {
                    log.error("DelayConsumer.onMessage error", var10);
                    throw new BizException("事件消息消费失败", var10);
                }
            }
        }
    
        /**
         * 消息级别，共6个
         * level-1 ：15s
         * level-2 : 30s
         * level-3 : 1m
         * level-4 : 5m
         * level-5 : 10m
         * level-6 : 30m
         */
        private Long getDelayTime(long timestamp, Integer level) {
            switch (level) {
                case 1:
                    return timestamp + 25 * 1000;
                case 2:
                    return timestamp + 30 * 1000;
                case 3:
                    return timestamp + 1 * 60 * 1000;
                case 4:
                    return timestamp + 5 * 60 * 1000;
                case 5:
                    return timestamp + 10 * 60 * 1000;
                case 6:
                    return timestamp + 30 * 60 * 1000;
            }
    
            return timestamp;
        }
    
    
        /**
         * 处理消息 并提交消息
         */
        private void processMessage(ConsumerRecord<String, String> record, Consumer consumer, TopicPartition topicPartition, String targetTopic, String msgKey, String event) {
    
            OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1);
            HashMap<TopicPartition, OffsetAndMetadata> metadataHashMap = new HashMap<>();
            metadataHashMap.put(topicPartition, offsetAndMetadata);
    
            //发布目标
            kafkaPublisher.publishSync(targetTopic, msgKey, event);
    
            log.info("DelayConsumer-records#offset={},targetTopic={},event={}", record.offset() + 1, targetTopic, event);
            consumer.commitSync(metadataHashMap);
        }
    
        /**
         * 重启消费
         */
        private void resume() {
            try {
                if (kafkaConsumer == null) {
                    return;
                }
    
                Set<TopicPartition> topicPartitions = kafkaConsumer.paused();
    
                if (CollectionUtil.isEmpty(topicPartitions)) {
                    return;
                }
    
                kafkaConsumer.resume(topicPartitions);
            } catch (Exception ex) {
                log.error("DelayConsumer-resume", ex);
            }
        }
    }

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

## 参考 ##

[RabbitMQ、RocketMQ、Kafka延迟队列实现-腾讯云开发者社区-腾讯云][RabbitMQ_RocketMQ_Kafka_-_-]

[延迟消息队列设计-腾讯云开发者社区-腾讯云][-_-]

[用Kafka实现延迟消息\_kafka延迟消费\_alvin.yao的博客-CSDN博客][Kafka_kafka_alvin.yao_-CSDN]

[怎么设计一个合适的延时队列？][Link 1]

[基于kafka实现延迟队列 - 知乎][kafka_ -]

[f5e9fe2ddfc2c2350ebad8a6d5362849.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/98b038e0a38740e991e26115201ee7c2.png
[be10947db4fbd86230c1da2b8c17712a.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/2007884777f749c9952a38d812a9afb2.png
[RabbitMQ_RocketMQ_Kafka_-_-]: https://cloud.tencent.com/developer/article/2225184
[-_-]: https://cloud.tencent.com/developer/article/1871910
[Kafka_kafka_alvin.yao_-CSDN]: https://blog.csdn.net/weixin_40270946/article/details/121293032
[Link 1]: https://mp.weixin.qq.com/s?__biz=MzU0OTE4MzYzMw==&mid=2247534953&idx=2&sn=256cca656ece217d72fd5fb93a0fe334&chksm=fbb1c097ccc649810eeb00d646bfbaafa7c2aeae7e9bf84aa59091faf5c75512442e7bdca357&scene=27
[kafka_ -]: https://zhuanlan.zhihu.com/p/365802989