redis和rabbitmq实现延时队列

redis和rabbitmq实现延时队列

  • 延迟队列使用场景
  • Redis中zset实现延时队列
  • Rabbitmq实现延迟队列

延迟队列使用场景

1. 订单超时处理
延迟队列可以用于处理订单超时问题。当用户下单后，将订单信息放入延迟队列，并设置一定的超时时间。如果在超时时间内用户未支付订单，消费者会从延迟队列中获取到该订单，并执行相应的处理操作，如取消订单、释放库存等。

2. 优惠券过期提醒
延迟队列可以用于优惠券的过期提醒功能。将即将过期的优惠券信息放入延迟队列，并设置合适的延迟时间。当延迟时间到达时，消费者将提醒用户优惠券即将过期，引导用户尽快使用。

3. 异步通知与提醒
延迟队列可以用于异步通知和提醒功能。例如，当用户完成某个操作后，系统可以将相关通知消息放入延迟队列，并设置一定的延迟时间，以便在合适的时机发送通知给用户。

Redis中zset实现延时队列

1. 创建延迟队列服务类

  • 创建一个延迟队列的服务类，例如DelayQueueService，用于操作Redis中的ZSet。这个服务类需要完成以下功能：
  • 将消息放入延迟队列：将消息作为元素添加到ZSet中，设置对应的延迟时间作为分数。轮询并处理已到期的消息：定时任务或者消息消费者轮询检查ZSet中的元素，获取到达指定时间的消息进行处理。删除已处理的消息：处理完消息后，从ZSet中将其删除。
@Service
public class DelayQueueService {private static final String DELAY_QUEUE_KEY = "delay_queue";@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void addToDelayQueue(String message,long delayTime){redisTemplate.opsForZSet().add(DELAY_QUEUE_KEY,message,System.currentTimeMillis()+delayTime);}public void processDelayedMessage(){//reverseRangeByScore 从高到低//rangeByScore 从低到高Set<String> messages = redisTemplate.opsForZSet().rangeByScore(DELAY_QUEUE_KEY, 0, System.currentTimeMillis());for(String message:messages){//处理消息System.out.println(message);redisTemplate.opsForZSet().remove(DELAY_QUEUE_KEY,message);}}
}

2. 配置定时任务或消息消费者
使用Spring Boot的定时任务或消息队列框架，定时调用延迟队列服务类的轮询方法或监听指定的消息队列,可以将轮训粒度放到1s一次。

@Component
public class DelayQueueSchedule {@Autowiredprivate DelayQueueService delayQueueService;// 每隔一段时间进行轮询并处理延迟消息@Scheduled(fixedDelay = 1000)public void pollAndProcessDelayedMessages() {delayQueueService.pollAndProcessDelayedMessages();}
}

然后在启动类上通过@EnableScheduling注解开启任务调度能力。

缺点：
使用ZSET（有序集合，Sorted Set）来实现延迟任务调度（如订单超时取消）是一种有效的方法，但它也有一些缺点和限制：

  1. 内存消耗：ZSET 在Redis中是一个有序集合，它需要占用一定的内存来存储成员和分数。如果你需要存储大量的延迟任务，可能会导致内存消耗较大。这可能会对Redis服务器的性能和成本产生影响，特别是在大规模应用中。
  2. 不适用于大规模延迟任务：ZSET 可以处理相对较小数量的延迟任务，但当需要管理大规模延迟任务队列时，可能会导致性能下降。在这种情况下，需要考虑更高效的延迟队列解决方案，例如使用分布式消息队列。
  3. 无法动态修改延迟时间： 一旦将任务添加到ZSET中，你不能轻松地修改任务的延迟时间。如果需要在任务已经添加后更改延迟时间，可能需要复杂的操作。
  4. 没有重试机制：ZSET 只能用于一次性延迟任务，无法自动处理任务失败后的重试。如果任务在执行时失败，你需要自己实现重试逻辑。
  5. 没有持久化： Redis是内存数据库，如果Redis服务器重启或发生故障，已添加的延迟任务数据将丢失。虽然可以通过Redis持久化机制来部分解决这个问题，但仍然存在一定风险。
  6. 复杂性增加： 使用ZSET来管理延迟任务队列需要编写复杂的代码来处理任务的添加、检索和删除。这可能增加应用程序的复杂性。

Rabbitmq实现延迟队列

死信，顾名思义就是无法被消费的消息。一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致queu 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

列出2种实现方式。
（1）使用Time To Live(TTL) + Dead Letter Exchanges（DLX）死信队列组合实现延迟队列的效果。
（2）使用RabbitMQ官方延迟插件rabbitmq_delayed_message_exchange，实现延时队列效果。

由于TTL（生存时间）过期导致的死信，就是我们实现延迟队列的的方式。
我们需要声明如下形式的交互机和队列，以及对应的routing key，并进行绑定：
请添加图片描述
上图绑定的代码如下所示

@Configuration
public class DeadQueueConfig {//普通交换机及队列public static final String X_EXCHANGE = "X";public static final String QUEUE_A = "QA";public static final String QUEUE_B = "QB";//死信交换机及队列public static final String Y_DEAD_LETTER_EXCHANGE = "Y";public static final String DEAD_LETTER_QUEUE = "QD";//通用队列public static final String QUEUE_C = "QC";// 声明 xExchange@Bean("xExchange")public DirectExchange xExchange() {return new DirectExchange(X_EXCHANGE);}//声明队列 A ttl 为 10s 并绑定到对应的死信交换机@Bean("queueA")public Queue queueA() {Map<String, Object> args = new HashMap<>(3);//声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//声明当前队列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//声明队列的 TTLargs.put("x-message-ttl", 10000);return QueueBuilder.durable(QUEUE_A).withArguments(args).build();}//声明队列A绑定X交换机  路由为XA@Beanpublic Binding queueABingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueA).to(xExchange).with("XA");}//声明队列 B ttl 为 40s 并绑定到对应的死信交换机@Bean("queueB")public Queue queueB() {Map<String, Object> args = new HashMap<>(3);//声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//声明当前队列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");//声明队列的 TTLargs.put("x-message-ttl", 40000);return QueueBuilder.durable(QUEUE_B).withArguments(args).build();}//声明队列 B 绑定 X 交换机@Beanpublic Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queue1B).to(xExchange).with("XB");}//声明通用队列C 不设ttl，由消费者决定ttl@Bean("queueC")public Queue queueC() {Map<String, Object> args = new HashMap<>(3);//声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);//声明当前队列的死信路由 keyargs.put("x-dead-letter-routing-key", "YD");return QueueBuilder.durable(QUEUE_C).withArguments(args).build();}// 声明队列 C 绑定 X 交换机@Beanpublic Binding queuecBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queueC).to(xExchange).with("XC");}// 声明 死信队列交换机@Bean("yExchange")public DirectExchange yExchange() {return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);}//声明死信队列 QD@Bean("queueD")public Queue queueD() {return new Queue(DEAD_LETTER_QUEUE,true);}//声明死信队列 QD 绑定关系@Beanpublic Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange) {return BindingBuilder.bind(queueD).to(yExchange).with("YD");}}

其中，QD为死信队列。当QA和QB队列中的消息，达到设定的TTL（10s和40s）后，将进入指定的死信队列QD。该方法缺点就是一个TTL对应一个队列

其中的QC作为通用的队列，即在消费者处指定消息对应的TTL，TTL过期后转入死信队列。使用该通用队列可以避免每增加一个新的时间需求，就要新增一个队列的问题。但该方法由于队列先进先出的性质，会导致一定的问题：

即先发出一个TTL为10s的消息a，进入队列；再马上发出一个TTL为2s的消息b，进入队列。由于队列的性质，会在消息a的TTL结束后，a进入死信队列后，b才会进入死信队列。而不是根据TTL的时间，b比a先进入死信队列。

声明交换机、队列，并绑定成功后，编写死信队列消费者代码;

@Component
@Slf4j
public class DeadQueueConsumer {@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel) throws IOException {String msg = new String(message.getBody());log.info("当前时间：{},收到死信队列信息：{}", new Date().toString(), msg);}
}

在controller中编写生产者代码，进行测试：

    @Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/sendMsg/{message}")public String sendMsg(@PathVariable String message){log.info("当前时间：{},发送一条信息给两个 TTL 队列:{}", new Date(), message);rabbitTemplate.convertAndSend("X", "XA", "消息来自 ttl 为 10S 的队列: " + message);rabbitTemplate.convertAndSend("X", "XB", "消息来自 ttl 为 40S 的队列: " + message);return "finish";}

结果如图：请添加图片描述
测试通用队列QC的效果：

@GetMapping("/send/{message}/{ttlTime}")public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {correlationData.getMessageProperties().setExpiration(ttlTime);return correlationData;});log.info("当前时间：{},发送一条时长{}毫秒 TTL 信息给队列 C:{}", new Date(), ttlTime, message);}

结果如下图
请添加图片描述

可以看到， 两条消息几乎同时到达死信队列，因为TTL为2s的消息由于被堵在TTL为10s的消息后导致。

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

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

相关文章

网络编程套接字——实现简单的TCP网络程序

网络编程套接字——实现简单的TCP网络程序

目录 1、TCP socket API详解 socket()&#xff1a; bind()&#xff1a; listen(): accept(): connect(): 2、简易的TCP网络程序 TcpServer.hpp TcpClient.cc Main.cc Log.hpp ThreadPool.hpp Task.hpp Init.hpp Daemon.hpp dict.txt Makefile 1、TCP socket A…
阅读更多...
CRM软件推荐2024：五款顶级产品解析，助您找到最佳选项！

CRM软件推荐2024：五款顶级产品解析，助您找到最佳选项！

一天之计在于晨&#xff0c;一年之计在于春。 2024年&#xff0c;民营经济发展继续壮大&#xff0c;这对于各行各业而言都是一种机遇挑战。企业想要规范化客户管理&#xff0c;实现销售增长&#xff0c;CRM软件仍然是一个不错的选择。在数字化时代&#xff0c;企业数字化转型已…
阅读更多...
基于SpringBoot实现WebSocket实时通讯的服务端和客户端

基于SpringBoot实现WebSocket实时通讯的服务端和客户端

实现功能 服务端注册的客户端的列表&#xff1b;服务端向客户端发送广播消息&#xff1b;服务端向指定客户端发送消息&#xff1b;服务端向多个客户端发送消息&#xff1b;客户端给服务端发送消息&#xff1b; 效果&#xff1a; 环境 jdk&#xff1a;1.8 SpringBoot&#x…
阅读更多...
大模型+强化学习_通过强化学习对齐大模型和环境

大模型+强化学习_通过强化学习对齐大模型和环境

英文名称: True Knowledge Comes from Practice: Aligning LLMs with Embodied Environments via Reinforcement Learning 中文名称: 实践出真知&#xff1a;通过强化学习将LLMS与具体环境对齐 链接: https://arxiv.org/abs/2401.14151 代码: https://github.com/WeihaoTan/TWO…
阅读更多...
BI技巧丨个性化视觉对象

BI技巧丨个性化视觉对象

BOSS&#xff1a;那个&#xff0c;那个谁&#xff0c;最近用户反映了&#xff0c;说是你们做的报表不太行啊&#xff1f;&#xff01; 白茶&#xff1a;&#xff08;&#xff1f;&#xff1f;&#xff1f;&#xff09;老板&#xff0c;怎么说&#xff1f; BOSS&#xff1a;就是…
阅读更多...
网络原理(5)——IP协议（网络层）

网络原理(5)——IP协议（网络层）

目录 一、IP协议报头介绍 1、4位版本 2、4位首部长度 3、8位服务器类型 4、16位总长度 5、16位标识位 6、3位标志位 7、13位偏移量 8、8位生存空间 9、8位协议 10、16位首部检验和 11、32位源IP地址 12、32位目的IP地址 二、IP协议如何管理地址&#xff1f; 1、动…
阅读更多...
Redis入门到实战-第三弹

Redis入门到实战-第三弹

Redis入门到实战 Redis数据类型官网地址Redis概述Redis数据类型介绍更新计划 Redis数据类型 官网地址 声明: 由于操作系统, 版本更新等原因, 文章所列内容不一定100%复现, 还要以官方信息为准 https://redis.io/Redis概述 Redis是一个开源的&#xff08;采用BSD许可证&#…
阅读更多...
《优化接口设计的思路》系列：第九篇—用好缓存，让你的接口速度飞起来

《优化接口设计的思路》系列：第九篇—用好缓存，让你的接口速度飞起来

一、前言 大家好&#xff01;我是sum墨&#xff0c;一个一线的底层码农&#xff0c;平时喜欢研究和思考一些技术相关的问题并整理成文&#xff0c;限于本人水平&#xff0c;如果文章和代码有表述不当之处&#xff0c;还请不吝赐教。 作为一名从业已达六年的老码农&#xff0c…
阅读更多...
2024Python计算机二级13

2024Python计算机二级13

一维数据采用线性方式组织&#xff0c;集合不属于线性结构 程调度仅负责对CPU进行分配 快速排序每经过一次元素的交换会产生新的逆序 记住队列中为一个元素的情况 区分二叉树与完全二叉树&#xff0c;前序序列是先访问根节点再访问左子树和右子树&#xff0c;中序序列是访问左子…
阅读更多...
《定时执行专家》：Nircmd 的超级搭档，解锁自动化新境界

《定时执行专家》：Nircmd 的超级搭档，解锁自动化新境界

目录 Nircmd 简介 《定时执行专家》与 Nircmd 的结合 示例&#xff1a; 自动清理电脑垃圾: 定时发送邮件: 定时关闭电脑: 《定时执行专家》的优势: 总结: 以下是一些其他使用示例&#xff1a; 立即下载《定时执行专家》&#xff1a; Nircmd 官方网站&#xff1a; 更…
阅读更多...
【数字IC/FPGA】书籍推荐（1）----《轻松成为设计高手--Verilog HDL实用精解》

【数字IC/FPGA】书籍推荐（1）----《轻松成为设计高手--Verilog HDL实用精解》

在下这几年关于数字电路、Verilog、FPGA和IC方面的书前前后后都读了不少&#xff0c;发现了不少好书&#xff0c;也在一些废话书上浪费过时间。接下来会写一系列文章&#xff0c;把一部分读过的书做个测评&#xff0c;根据个人标准按十分制满分来打分分享给大家。 书名&#xf…
阅读更多...
企业微信可以更换公司主体吗？

企业微信可以更换公司主体吗？

企业微信变更主体有什么作用&#xff1f;当我们的企业因为各种原因需要注销或已经注销&#xff0c;或者运营变更等情况&#xff0c;企业微信无法继续使用原主体继续使用时&#xff0c;可以申请企业主体变更&#xff0c;变更为新的主体。企业微信变更主体的条件有哪些&#xff1…
阅读更多...
基于Docker的JMeter分布式压测！

基于Docker的JMeter分布式压测！

一个JMeter实例可能无法产生足够的负载来对你的应用程序进行压力测试。如本网站所示&#xff0c;一个JMeter实例将能够控制许多其他的远程JMeter实例&#xff0c;并对你的应用程序产生更大的负载。JMeter使用Java RMI[远程方法调用]来与分布式网络中的对象进行交互。JMeter主站…
阅读更多...
Vue js封装接口

Vue js封装接口

天梦星服务平台 (tmxkj.top)https://tmxkj.top/#/ 1.安装axios npm install axios -g 2.在src下新建一个Api文件夹,再创建一个js文件 import axios from axios let configuration {url:"http://localhost:9090" } /*** 请求项目数据的请求体*/ async function h…
阅读更多...
CV论文--2024.3.20

CV论文--2024.3.20

1、Graph Expansion in Pruned Recurrent Neural Network Layers Preserve Performance 中文标题&#xff1a;图扩展在修剪的循环神经网络层中保持性能 简介&#xff1a;这段摘要讨论了图的扩展性质&#xff0c;包括强连通性和稀疏性。研究表明&#xff0c;深度神经网络可以通…
阅读更多...
linux -- I2C设备驱动 -- MS32006（低压5V多通道电机驱动器）

linux -- I2C设备驱动 -- MS32006（低压5V多通道电机驱动器）

产品简述 MS32006 是一款多通道电机驱动芯片， 其中包含两路步进电机驱动， 一路直流电机驱动； 每个通道的电流最高电流1.0A； 支持两相四线与四相五线步进电机。芯片采用 I2C 的通信接口控制模式， 兼容 3.3V/5V 的标准工业接口。 MS32006 总共集成了两路步进电机驱动器与一…
阅读更多...
【c++入门】引用，内联函数，auto

【c++入门】引用，内联函数，auto

&#x1f525;个人主页&#xff1a;Quitecoder &#x1f525;专栏&#xff1a;c笔记仓 朋友们大家好&#xff0c;本节我们来到c中一个重要的部分&#xff1a;引用 目录 1.引用的基本概念与用法1.1引用特性1.2使用场景1.3传值、传引用效率比较1.4引用做返回值1.5引用和指针的对…
阅读更多...
公司调研 | 空间机械臂GITAI | 日企迁美

公司调研 | 空间机械臂GITAI | 日企迁美

最近做的一些公司 / 产品调研没有从技术角度出发&#xff0c;而更关注宏观发展&#xff1a;主营方向、产品介绍、商业化落地情况、融资历程、公司愿景、创始人背景等。部分调研放在知乎上&#xff0c;大部分在飞书私人链接上 最近较关注人形Robot的发展情况&#xff0c;欢迎感兴…
阅读更多...
【Java】Map和Set

【Java】Map和Set

文章目录 一、Map和Set的概念二、模型三、Map的说明3.1 Map.Entry<K, V>的说明3.2 Map 的常用方法 四、Set的说明4.1 Set的常用方法 一、Map和Set的概念 Map和set是一种专门用来进行搜索的容器或者数据结构&#xff0c;其搜索的效率与其具体的实例化子类有关&#xff0c…
阅读更多...
在线播放视频网站源码系统 带完整的安装代码包以及搭建教程

在线播放视频网站源码系统 带完整的安装代码包以及搭建教程

在线播放视频网站源码系统的开发&#xff0c;源于对当前视频市场的深入洞察和用户需求的精准把握。随着视频内容的爆炸式增长&#xff0c;用户对视频播放的需求也日益多样化。他们希望能够随时随地观看自己感兴趣的视频内容&#xff0c;同时还希望能够在观看过程中享受到流畅、…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023