目录
前言
一、延时队列实用场景
二、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);}}
}
参考
RabbitMQ、RocketMQ、Kafka延迟队列实现-腾讯云开发者社区-腾讯云
延迟消息队列设计-腾讯云开发者社区-腾讯云
用Kafka实现延迟消息_kafka延迟消费_alvin.yao的博客-CSDN博客
怎么设计一个合适的延时队列?
基于kafka实现延迟队列 - 知乎