前言

一、延时队列实用场景

二、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 {@Getterprivate UserRegisterMessage message;private long delayTime;public UserDelayTask(UserRegisterMessage message, long delayTime) {this.message = message;// 延迟时间加当前时间this.delayTime = System.currentTimeMillis() + delayTime;}// 获取任务剩余时间@Overridepublic long getDelay(TimeUnit unit) {return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}@Overridepublic 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);

开启线程处理延迟任务

 @Overridepublic 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 中。

四、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。

实现

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>>() {@Overridepublic void onSuccess(SendResult<String, String> result) {//发送成功后回调log.info("{}-异步发送成功, result={}。", topic, result.getRecordMetadata().toString());}@Overridepublic void onFailure(Throwable throwable) {//发送失败回调log.error("{}-异步发送失败。", topic, throwable);}});} catch (Exception ex) {log.error("{}-异步发送异常。", topic, ex);}}

消费者代码

/*** 参考RocketMQ支持延迟消息设计，不支持任意时间精度的延迟消息，只支持特定级别的延迟消息，* 将消息延迟等级分为1s、5s、10s 、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h，共18个级别，* 只创建一个有18个分区的延时topic，每个分区对应不同延时等级。** https://blog.csdn.net/weixin_40270946/article/details/121293032** 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;/*** 消费者*/@Resource(name = "kafkaConsumer")private KafkaConsumer<String, String> kafkaConsumer;/*** 延迟消息发布*/@Resourceprivate DelayMessagePublisher delayMessagePublisher;@PostConstructpublic void init() {ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();//当系统需要循环间隔一定时间执行某项任务的时候可以使用scheduleWithFixedDelay方法来实现executorService.scheduleWithFixedDelay(() -> {synchronized (lock) {resume();lock.notifyAll();log.info("DelayConsumer-notifyAll");}}, 0, interval, TimeUnit.MILLISECONDS);}/*** 批量消费消息*/@KafkaListener(topics = "#{'${delayTopic.topic}'}", groupId = "#{'${spring.kafka.consumer.group-id}'}")public void onMessage(List<ConsumerRecord<String, String>> records, Consumer consumer) {synchronized (lock) {try {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"));TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());long delayTime = getDelayTime(timestamp, level);if (delayTime <= System.currentTimeMillis()) {log.info("DelayConsumer-delayTime={} <= currentTime={}", delayTime, System.currentTimeMillis());// 处理消息processMessage(record, consumer, topicPartition, targetTopic, event);} else {log.info("DelayConsumer-delayTime={} > currentTime={}", delayTime, System.currentTimeMillis());// 暂停消费consumer.pause(Collections.singletonList(topicPartition));consumer.seek(topicPartition, record.offset());delay = true;break;}}if (delay) {lock.wait();}} catch (Exception var10) {log.error("{}.onMessage#error . message={}");throw new BizException("事件消息消费失败", var10);}}}/*** 消息级别，共18个* level-1 ：30s* level-2 : 1m* level-3 : 5m* level-4 : 10m* level-5 : 20m* level-6 : 30m*/private Long getDelayTime(long timestamp, Integer level) {switch (level) {case 1:return timestamp + 1 * 1000;case 2:return timestamp + 5 * 1000;case 3:return timestamp + 10 * 1000;case 4:return timestamp + 30 * 1000;case 5:return timestamp + 1 * 60 * 1000;case 6:return timestamp + 2 * 60 * 1000;//.........  省略}return timestamp;}/*** 处理消息 并提交消息*/private void processMessage(ConsumerRecord<String, String> record, Consumer consumer, TopicPartition topicPartition, String targetTopic, String event) {OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1);HashMap<TopicPartition, OffsetAndMetadata> metadataHashMap = new HashMap<>();metadataHashMap.put(topicPartition, offsetAndMetadata);delayMessagePublisher.sendMessage(targetTopic, event);log.info("DelayConsumer-records#offset={},targetTopic={},event={}", record.offset() + 1, targetTopic, event);consumer.commitSync(metadataHashMap);}/*** 重启消费*/private void resume() {try {kafkaConsumer.resume(kafkaConsumer.paused());} catch (Exception ex) {log.error("DelayConsumer-resume", ex);}}
}