延时消息队列

目录

前言

一、延时队列实用场景

二、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到当前ConsumerRecordoffset位置,同时使用定时器去轮询延迟的TopicPartition,达到延迟时间之后进行resume。

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

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

这里为什么要进行pauseresume呢?因为如果不这样的话,如果超时未消费达到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实现延迟队列 - 知乎

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/104069.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Python函数的概念以及定义方式

一. 前言 嗨喽~大家好呀&#xff0c;这里是魔王呐 ❤ ~! python更多源码/资料/解答/教程等 点击此处跳转文末名片免费获取 二. 什么是函数&#xff1f; 假设你现在是一个工人&#xff0c;如果你实现就准备好了工具&#xff0c;等你接收到任务的时候&#xff0c; 直接带上工…

Spring初始化项目

1、官网用法 访问地址&#xff1a;https://start.spring.io idea配置&#xff1a;https://start.spring.io 2、阿里巴巴加速 访问地址&#xff1a;https://start.aliyun.com/bootstrap.html idea配置&#xff1a;https://start.aliyun.com 3、区别 官网阿里巴巴版本最新稍…

2023区块链应用操作员认证(4级)报名来弘博创新

区块链应用操作员&#xff0c;是指运用区块链技术及工具&#xff0c;从事政务、金融、医疗、教育、养老等场景系统应用操作的人员。 腾讯作为广东省第一批公布的社会培训评价组织&#xff0c;可开展职业技能等级认定职业(工种)区块链应用操作员(4-3-2-1级)。 证书含金量 证书是…

Redis 集群

1. 是什么 1.1 定义 由于数据量过大&#xff0c;单个Master复制集难以承担&#xff0c;因此需要对多个复制集进行集群&#xff0c;形成水平扩展每个复制集只负责存储整个数据集 的一部分&#xff0c;这就是Redis的集群&#xff0c;其作用是提供在多个Redis节点间共享数据的程序…

如何为虚拟机添加磁盘,扩充原有分区的磁盘空间

如何为虚拟机添加磁盘&#xff0c;扩充原有分区的磁盘空间 关机新增磁盘 虚拟机关机的状态下&#xff0c;在 VMware 当中新增一块磁盘&#xff0c;选中左边要添加磁盘的虚拟机镜像&#xff0c;然后鼠标右键点击设置。 选中磁盘点击添加 点击下一步&#xff0c;悬着SCSI这个…

【C++基础】7. 控制语句

文章目录 【 1. 循环 】1.1 循环类型1.2 循环控制语句break 语句continue 语句goto 语句 1.3 无限循环 【 2. 选择 】switch 语句&#xff1f;&#xff1a;语句 【 1. 循环 】 1.1 循环类型 循环类型描述while 循环当给定条件为真时&#xff0c;重复语句或语句组。它会在执行…

人体呼吸存在传感器成品,毫米波雷达探测感知技术,引领智能家居新潮流

随着科技的不断进步和人们生活质量的提高&#xff0c;智能化家居逐渐成为一种时尚和生活方式。 人体存在传感器作为智能家居中的重要组成部分&#xff0c;能够实时监测环境中人体是否存在&#xff0c;为智能家居系统提供更加精准的控制和联动。 在这个充满创新的时代&#xf…

软件测试中的43个功能测试点总结

功能测试就是对产品的各功能进行验证&#xff0c;根据功能测试用例&#xff0c;逐项测试&#xff0c;检查产品是否达到用户要求的功能。针对web系统的常用测试方法如下&#xff1a; 1、页面链接检查&#xff1a; 每一个链接是否都有对应的页面&#xff0c;并且页面之间切换正…

EagleSDR USB HAT FT600

给EagleSDR做了个USB 3.0的子卡&#xff0c;采用FT600方案&#xff0c;实物如下&#xff1a; 用FT600DataStreamerDemoApp测试&#xff0c;速度如下&#xff1a; 由于FT600是16bit的接口&#xff0c;如果用FT601的32bit接口&#xff0c;性能应该还会有大幅提升。 测试代码很简…

LeetCode 297. Serialize and Deserialize Binary Tree【树,DFS,BFS,设计,二叉树,字符串】困难

本文属于「征服LeetCode」系列文章之一&#xff0c;这一系列正式开始于2021/08/12。由于LeetCode上部分题目有锁&#xff0c;本系列将至少持续到刷完所有无锁题之日为止&#xff1b;由于LeetCode还在不断地创建新题&#xff0c;本系列的终止日期可能是永远。在这一系列刷题文章…

使用动态住宅代理还能带来哪些好处?

一、什么是动态住宅代理ip 动态住宅代理是一种代理技术&#xff0c;它利用代理服务器中转用户和目标服务器之间的网络流量&#xff0c;实现用户真实位置的屏蔽。代理提供商会有自己的ip大池子&#xff0c;当你通过代理服务器向网站发送请求时&#xff0c;服务器会从池子中选中…