RabbitMQ 延时消息实现

1. 实现方式

1. 设置队列过期时间：延迟队列消息过期 + 死信队列，所有消息过期时间一致
2. 设置消息的过期时间：此种方式下有缺陷，MQ只会判断队列第一条消息是否过期，会导致消息的阻塞需要额外安装 `rabbitmq_delayed_message_exchange` 插件才能解决此问题
  • 导入Spring 集成RabbitMQ MAEVN
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.2.5.RELEASE</version>
</dependency>

2. 设置队列过期时间：延迟队列消息过期 + 死信队列

推送消息至延迟队列 -> 消息过期自动推送到死信队列 -> 消费死信队列

2.1. MQ配置信息

2.1.1. 自定义队列配置

…/bootstrap.yml

# rabbitmq自定义配置
rabbitmq:ttlExchange: medical_dev_ttl_topic_changettlKey: dev_ttlttlQueue: medical.dev.ttl.topic.queuedelayExpireTime: 600ttlQueueSize: 10000deadExchange: medical_dev_dead_topic_changedeadKey: dev_deaddeadQueue: medical.dev.dead.topic.queue
2.1.2. 读取自定义MQ配置信息
/*** amqp配置文件*/
@Data
@Component
@ConfigurationProperties(prefix = "rabbitmq")
public class MyConfigProperties {/*** 延迟队列*/public String ttlExchange;public String ttlKey;public String ttlQueue;private Integer delayExpireTime;public Integer ttlQueueSize;/*** 死信队列*/public String deadExchange;public String deadKey;public String deadQueue;}

2.2. 配置文件自动生成队列

2.2.1. 延迟队列
import com.awsa.site.mq.MyConfigProperties;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.Resource;
import java.util.HashMap;/*** 延迟队列配置文件* * @author mingAn.xie*/
@Configuration
public class RabbitMQConfigTTL {@ResourceMyConfigProperties myConfigProperties;// 1: 声明交换机@Beanpublic TopicExchange ttlTopicExchange(){return new TopicExchange(myConfigProperties.getTtlExchange());}// 2: 声明队列// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。@Beanpublic Queue ttlTopicduanxinQueue(){HashMap<String, Object> args = new HashMap<>();// 给队列设置消息过期时间：毫秒值args.put("x-message-ttl", mqConfigProperties.getDelayExpireTime() * 1000);// 设置队列最大长度args.put("x-max-length", myConfigProperties.getTtlQueueSize());// 设置死信队列交换机名称// 当消息在一个队列中变成死信后,它能就发送到另一个交换机中，这个交换机就是DLX，绑定DLX的队列被称之为死信队列// 编程死信队列的原因：消息被拒绝，消息过期，队列达到最大长度args.put("x-dead-letter-exchange", myConfigProperties.getDeadExchange());// 设置死信队列路由keyargs.put("x-dead-letter-routing-key", myConfigProperties.getDeadKey());return new Queue(myConfigProperties.getTtlQueue(), true, false, false, args);}// 3: 绑定对用关系@Beanpublic Binding ttlTopicsmsBinding(){return BindingBuilder.bind(ttlTopicduanxinQueue()).to(ttlTopicExchange()).with(myConfigProperties.getTtlKey());}}
2.2.2. 死信队列

import com.awsa.site.mq.MyConfigProperties;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.Resource;/*** 死信队列配置文件* * @author mingAn.xie*/
@Configuration
public class RabbitMQConfigDead {@ResourceMyConfigProperties myConfigProperties;// 1: 声明交换机@Beanpublic TopicExchange deadTopicExchange(){return new TopicExchange(myConfigProperties.getDeadExchange());}// 2: 声明队列// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。@Beanpublic Queue deadTopicduanxinQueue(){return new Queue(myConfigProperties.getDeadQueue(), true);}// 3: 绑定对用关系@Beanpublic Binding deadTopicsmsBinding(){return BindingBuilder.bind(deadTopicduanxinQueue()).to(deadTopicExchange()).with(myConfigProperties.getDeadKey());}}

2.3. 生产者推送消息

import com.awsa.site.mq.MyConfigProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;/*** RabbitMQ生产者推送消息类* * @author xiemingan*/
@Component
@Slf4j
public class RabbitmqProducer {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyConfigProperties myConfigProperties;/*** @param pushMessage 推送消息体*/public void pushTtlMessage(String pushMessage) {// 推送消息至交换机，并指定路由keyrabbitTemplate.convertAndSend(myConfigProperties.getTtlExchange(), myConfigProperties.getTtlKey(), pushMessage);log.info("MQ消息推送队列, exchange: {}, key: {}, message: {}", myConfigProperties.getTtlExchange(), myConfigProperties.getTtlKey(), pushMessage);}}

2.4. 消费者处理消息

import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;/*** @author mingAn.xie*/
@Log4j2
@Component
public class RabbitmqConsumer {/*** 消费死信队列* @param message 消息体*/@RabbitListener(queues = "${rabbitmq.deadQueue}")public void pushMessages(Message message) {String body = new String(message.getBody()).trim();if (StringUtils.isEmpty(body)){return;}log.info("MQ消息消费, RabbitmqConsumer.pushMessages() : {}", body);}}

3. 设置消息的过期时间

设置交换机类型为 x-delayed-type，推送消息至交换机，直连队列消费

3.1. 安装插件 rabbitmq_delayed_message_exchange

前言：这里默认使用环境为 Liunx 系统 Docker 安装 RabbitMQ

具体可以参考这篇文章：Docker 安装 RabbitMQ 挂载配置文件

安装插件版本需要与RabbitMQ版本一致，否则可能会导致安装失败，可先进入RabbitMQ容器中查看其他插件版本

插件各版本地址: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

  • 这里以最新版本 v3.13.0 举例
# 下载插件
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.13.0/rabbitmq_delayed_message_exchange-3.13.0.ez# 将插件复制进容器中: rabbitmq_xxxxxx
docker cp rabbitmq_delayed_message_exchange-3.13.0.ez rabbitmq_xxxxxx:/plugins# 进入容器: rabbitmq_xxxxxx
docker exec -it rabbitmq_xxxxxx bash
cd plugins# 查询插件列表, 此处可看到插件的版本
rabbitmq-plugins list# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  • 交换机类型中出现 x-delayed-type 表示安装成功

3.2. MQ配置信息

3.2.1. 自定义队列配置

…/bootstrap.yml

#mq队列自定义配置
rabbitmq:saveTaskTtlExchange: ey240001_pro_save_task_ttl_topic_exchangesaveTaskTtlKey: ey240001_pro_save_task_ttlsaveTaskTtlQueue: ey240001.pro.save.task.ttl.topic.queuesaveTaskTtlQueueSize: 10000
3.2.2. 读取自定义MQ配置信息
/*** amqp配置文件** @author mingAn.xie*/
@Data
@Component
@ConfigurationProperties(prefix = "rabbitmq")
public class MyConfigProperties {/*** 任务待办生成延时队列*/public String saveTaskTtlExchange;public String saveTaskTtlKey;public String saveTaskTtlQueue;public Integer saveTaskTtlQueueSize;}

3.3. 配置文件生成 x-delayed-type 交换机

import com.awsa.site.mq.MyConfigProperties;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;/*** x-delayed-type 交换机延迟队列配置* * @author mingAn.xie*/
@Configuration
public class RabbitMQConfigSaveTaskTtl {@ResourceMyConfigProperties myConfigProperties;// 1: 声明交换机@Beanpublic CustomExchange saveTaskTopicExchange() {Map<String, Object> args = new HashMap<>();// 设置延迟队列插件类型：按过期时间消费args.put("x-delayed-type", "direct");// 参数：name 交换机名称，type 交换机类型，durable 是否持久化，autoDelete 是否自动删除，arguments 参数return new CustomExchange(myConfigProperties.getSaveTaskTtlExchange(), "x-delayed-message", true, false, args);}// 2: 声明队列// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。@Beanpublic Queue saveTaskTopicduanxinQueue() {return new Queue(myConfigProperties.getSaveTaskTtlQueue(), true, false, false);}// 3: 绑定对用关系@Beanpublic Binding saveTaskTopicsmsBinding() {return BindingBuilder.bind(saveTaskTopicduanxinQueue()).to(saveTaskTopicExchange()).with(myConfigProperties.getSaveTaskTtlKey()).noargs();}}

3.4. 生产者推送消息

import com.awsa.site.mq.MyConfigProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;/*** 生产者推送消息类* * @author xiemingan*/
@Component
@Slf4j
public class RabbitmqProducer {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyConfigProperties myConfigProperties;/*** @param pushMessage 推送消息体* @param ttlTime     延时时间（毫秒值）*/public void pushTtlMessage(String pushMessage, long ttlTime) {ttlTime = ttlTime <= 0 ? 1000 : ttlTime;// 3.1.推送MQ延迟消息队列long finalTtlTime = ttlTime;MessagePostProcessor messagePostProcessor = message -> {// 设置延迟时间message.getMessageProperties().setDelay((int) finalTtlTime);return message;};rabbitTemplate.convertAndSend(myConfigProperties.getSaveTaskTtlExchange(), myConfigProperties.getSaveTaskTtlKey(), pushMessage, messagePostProcessor);log.info("MQ消息推送队列, exchange: {}, key: {}, message: {}, ttlTime: {}", myConfigProperties.getSaveTaskTtlExchange(), myConfigProperties.getSaveTaskTtlKey(), pushMessage, ttlTime);}}

3.5. 消费者处理消息

import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;/*** @author mingAn.xie*/
@Log4j2
@Component
public class RabbitmqConsumer {/*** 消费延时消息* @param message 消息体*/@RabbitListener(queues = "${rabbitmq.saveTaskTtlQueue}")public void pushMessages(Message message) {String body = new String(message.getBody()).trim();if (StringUtils.isEmpty(body)) {return;}log.info("MQ延迟消息消费, RabbitmqConsumer.pushMessages() : {}", body);}}

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

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

相关文章

光伏发电量预测（Python代码，CNN结合LSTM，TensorFlow框架）

光伏发电量预测（Python代码，CNN结合LSTM，TensorFlow框架）

1.数据集&#xff08;开始位置&#xff09;&#xff0c;数据集免费下载链接&#xff1a;https://download.csdn.net/download/qq_40840797/89051099 数据集一共8列&#xff0c;第一列是时间&#xff0c;特征列一共有6列&#xff1a;"WindSpeed" - 风速 "Sunshi…
阅读更多...
C++template之类模版进一步了解

C++template之类模版进一步了解

前言&#xff1a;这一篇是在我的上一篇文章的基础上&#xff0c;再进一步所写的。 链接&#xff1a;CTemplate&#xff1c;&#xff1e;模版的介绍及深度解析-CSDN博客 一、类模板实例化 1.非类型模版参数 类型模版参数&#xff1a;就是跟在 class后面或者typename后的类型 非…
阅读更多...
visual studio报：引发的异常:“System.DllNotFoundException”(位于 ConsoleCAN1.exe 中)

visual studio报：引发的异常:“System.DllNotFoundException”(位于 ConsoleCAN1.exe 中)

最近在重构CAN通信的代码&#xff0c;把论文中的java转为C#实现&#xff0c;由于某种原因&#xff0c;java不能复现&#xff0c;所以转为c#。 然而c#的重构过程遇到许多问题&#xff0c;因为两种语言的编程方式、线程等等实现上有所差异。 其中一个错误&#xff1a; 引发的异…
阅读更多...
vue3+threejs新手从零开发卡牌游戏（二十一）：添加战斗与生命值关联逻辑

vue3+threejs新手从零开发卡牌游戏（二十一）：添加战斗与生命值关联逻辑

首先将双方玩家的HP存入store中&#xff0c;stores/common.ts代码如下&#xff1a; import { ref, computed } from vue import { defineStore } from piniaexport const useCommonStore defineStore(common, () > {const _font ref() // 字体const p1HP ref(4000) // 己…
阅读更多...
三菱Q系列PLC以太网TCP通讯FB块源码

三菱Q系列PLC以太网TCP通讯FB块源码

三菱Q系列PLC的tcp通讯&#xff0c;客户端和服务器两个变量好用的FB块&#xff0c;调用块就可以实现通讯连接&#xff0c;不需要自己写程序&#xff0c;简单配置引脚就可以。该块还集成了断网&#xff0c;连接错误&#xff0c;发送接收数据错误报警等功能。具体功能见下面介绍.…
阅读更多...
sql Tuning Advisor启用导致业务性能问题

sql Tuning Advisor启用导致业务性能问题

数据库每天晚上10点后业务性能很卡&#xff0c;大量的insert被堵塞&#xff0c;查询等待事件发现有大量的“library cache lock”和“cursor: pin S wait on X”。 22:00数据库的统计信息开始收集&#xff0c; Sql Tuning Advisor堵塞了统计信息的收集&#xff0c;等待事件是“…
阅读更多...
利用HIVE的窗口函数进行SQL查询中出现的问题记录

利用HIVE的窗口函数进行SQL查询中出现的问题记录

student_info部分数据 score_info部分数据 course_info 1、问题复现 --完整SQL selectsti.stu_id,sti.stu_name,concat_ws(",",collect_set(ci.course_name)) over(partition by sti.stu_id) fromstudent_info sti left joinscore_info sci onsti.stu_idsci.stu_id l…
阅读更多...
mysql之MyBatis核心工作原理

mysql之MyBatis核心工作原理

MyBatis核心工作原理 一、源码环境 1.手动编译源码 工欲善其事必先利其器。为了方便我们在看源码的过程中能够方便的添加注释&#xff0c;我们可以自己来从官网下载源码编译生成对应的Jar包&#xff0c;然后上传到本地maven仓库&#xff0c;再引用这个Jar。大家可以自行去官…
阅读更多...
matlab 点云可视化(6)——点云按强度进行可视化

matlab 点云可视化(6)——点云按强度进行可视化

目录 一、功能概述1、算法概述2、主要函数二、代码示例三、结果展示四、参考链接本文由CSDN点云侠原创原文链接。如果你不是在点云侠的博客中看到该文章，那么此处便是不要脸的爬虫。 一、功能概述 1、算法概述 点云按强度进行可视化 2、主要函数
阅读更多...
绿联 安装Uptime Kuma - 一款开源的服务器监控和状态检测工具

绿联 安装Uptime Kuma - 一款开源的服务器监控和状态检测工具

Uptime Kuma 功能简介 Uptime Kuma 是一款开源的服务器监控和状态检测工具&#xff0c;它帮助您跟踪服务器的可用性、性能和健康状态。 主要功能&#xff1a; 服务器监控 Uptime Kuma 可以监控多个服务器&#xff0c;包括 Web 服务器、数据库服务器、应用程序服务器等。 它会定…
阅读更多...
Antd Vue3 使用 Anchor 锚点组件记录

Antd Vue3 使用 Anchor 锚点组件记录

项目场景 客户要求做一个表单页面&#xff0c;表单数据分为三步&#xff0c;每一步骤是一个单独的 Vue 组件&#xff0c;表单上方需要使用锚点组件实现锚点定位到每一步的功能。 代码总览 <template><div class"guided-form-content-wrapper"><!-- …
阅读更多...
初步了解C++

初步了解C++

目录 一&#xff1a;什么是C&#xff1f; 二.C发展史 三:C关键字 四&#xff1a;命名空间 4.1命名空间的介绍 4.2命名空间的使用 4.3命名空间的使用 4.3.1使用作用域限定符 4.3.2 使用using将命名空间的某个成员引入 4.3.3使用using把整个命名空间展开 4.4命名空…
阅读更多...
每天学习一个Linux命令之scp

每天学习一个Linux命令之scp

每天学习一个Linux命令之scp 在Linux系统中&#xff0c;scp&#xff08;Secure Copy&#xff09;是一个用于在本地主机和远程主机之间进行文件传输的命令行工具。它基于SSH协议&#xff0c;通过加密方式传输文件&#xff0c;确保传输的安全性和完整性。scp命令非常强大且使用简…
阅读更多...
【蓝桥杯选拔赛真题50】C++简易炸弹超人 第十四届蓝桥杯青少年创意编程大赛 算法思维 C++编程选拔赛真题解析

【蓝桥杯选拔赛真题50】C++简易炸弹超人 第十四届蓝桥杯青少年创意编程大赛 算法思维 C++编程选拔赛真题解析

目录 C简易炸弹超人 一、题目要求 1、编程实现 2、输入输出 二、算法分析 三、程序编写 四、程序说明 五、运行结果 六、考点分析 七、推荐资料 C简易炸弹超人 第十四届蓝桥杯青少年创意编程大赛C选拔赛真题 一、题目要求 1、编程实现 有一块矩形游戏场地&#x…
阅读更多...
聊聊k8s服务发现的优缺点

聊聊k8s服务发现的优缺点

序 本文主要研究一下使用k8s服务发现的优缺点 spring cloud vs kubernetes 这里有张spring cloud与kubernetes的对比&#xff0c;如果将微服务部署到kubernetes上面&#xff0c;二者有不少功能是重复的&#xff0c;可否精简。 这里主要是讲述一下如果不使用独立的服务发现&am…
阅读更多...
SpringMVC注解及使用规则

SpringMVC注解及使用规则

文章目录 前言一、SpringMVC注解是什么&#xff1f;二、使用步骤1.注解使用2创建JSP3 SpringMVC视图1. 逻辑视图&#xff08;Logical View&#xff09;2. 物理视图&#xff08;Physical View&#xff09;区别和关系 4 SpringMVC注解总结 总结 前言 提示&#xff1a;这里可以添…
阅读更多...
建立一个简单的网页音乐盒模型效果#css#h5

建立一个简单的网页音乐盒模型效果#css#h5

“音乐盒”可以看做一个大盒子&#xff0c;用<div>标签进行定义。大盒子的上面为文本内容&#xff0c;可以在<div>标签中嵌套<h2>和<p>标签来实现&#xff1b;大盒子下面为图像&#xff0c;通过在<div>标签中嵌套<img/>标签来实现。 样式…
阅读更多...
[Python人工智能] 四十五.命名实体识别 (6)利用keras构建CNN-BiLSTM-ATT-CRF实体识别模型（注意力问题探讨）

[Python人工智能] 四十五.命名实体识别 (6)利用keras构建CNN-BiLSTM-ATT-CRF实体识别模型（注意力问题探讨）

从本专栏开始，作者正式研究Python深度学习、神经网络及人工智能相关知识。前文讲解融合Bert的实体识别研究，使用bert4keras和kears包来构建Bert+BiLSTM-CRF模型。这篇文章将详细结合如何利用keras和tensorflow构建基于注意力机制的CNN-BiLSTM-ATT-CRF模型，并实现中文实体识别…
阅读更多...
notepad++里安装32位和64位的16进制编辑器Hex-Editor

notepad++里安装32位和64位的16进制编辑器Hex-Editor

这个16进制编辑器确实是个好东西&#xff0c;平时工作种会经常用到&#xff0c; 这是hex-editor的官网。这个里边只能下载32位的(64位的看最下边)&#xff0c;选一个合适的版本&#xff0c;我当时选的是最新的版本 https://sourceforge.net/projects/npp-plugins/files/Hex%20E…
阅读更多...
ARP协议定义及工作原理

ARP协议定义及工作原理

ARP的定义 地址解析协议(Address Resolution Protocol&#xff0c;ARP)&#xff1a;ARP协议可以将IPv4地址(一种逻辑地址)转换为各种网络所需的硬件地址(一种物理地址)。换句话说&#xff0c;所谓的地址解析的目标就是发现逻辑地址与物理地址的映射关系。 ARP仅用于IPv4协议&a…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023