RabbitMQ 延时消息实现

news/2024/5/14 8:00:38/文章来源:https://blog.csdn.net/weixin_46759279/article/details/137142279

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.luyixian.cn/news_show_1029185.aspx

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

相关文章

【MySQL】DQL-条件查询语句全解（附带代码演示＆案例练习）

【MySQL】DQL-条件查询语句全解（附带代码演示＆案例练习）

前言 大家好吖&#xff0c;欢迎来到 YY 滴MySQL系列 &#xff0c;热烈欢迎&#xff01; 本章主要内容面向接触过C Linux的老铁 主要内容含&#xff1a; 欢迎订阅 YY滴C专栏&#xff01;更多干货持续更新&#xff01;以下是传送门&#xff01; YY的《C》专栏YY的《C11》专栏YY的…
阅读更多...
C++中的STL简介与string类

C++中的STL简介与string类

目录 STL简介 STL的版本 STL的六大组件 string类 标准库中的string类 string类的常用接口 string类对象对容量的操作 size()函数与length()函数 capacity()函数 capacity的扩容方式 reserve()函数 resize()函数 string类对象的操作 push_back()函数 append()函数 operator()函数…
阅读更多...
Rust高级爬虫：如何利用Rust抓取精美图片

Rust高级爬虫：如何利用Rust抓取精美图片

引言 在当今信息爆炸的时代&#xff0c;互联网上的图片资源丰富多彩&#xff0c;而利用爬虫技术获取这些图片已成为许多开发者的关注焦点。本文将介绍如何利用Rust语言进行高级爬虫编程&#xff0c;从而掌握抓取精美图片的关键技术要点。 Rust爬虫框架介绍 Rust语言生态中有…
阅读更多...
【谷歌开发者月刊】聚焦三月精彩内容，让开发思路更加开阔

【谷歌开发者月刊】聚焦三月精彩内容，让开发思路更加开阔

随着春日的到来&#xff0c;阳光渐煦&#xff0c;正是吸收能量的大好时机&#xff0c;我们也为开发者们带来了众多更新内容&#xff0c;为您的开发之路提供思路&#xff01;本月精彩内容众多&#xff0c;快来一起查收&#xff01; 本月看点 01Android 15 首个开发者预览版到来0…
阅读更多...
git基本操作（小白入门快速上手一）

git基本操作（小白入门快速上手一）

1、前言 我们接上一篇文章来讲&#xff0c;直接开干 1.1、工作区 1. 工作区很好理解&#xff0c;就是我们能看到的工作目录&#xff0c;就是本地的文件夹。 2. 这些本地的文件夹我们要通过 git add 命令先将他们添加到暂存区中。 3. git commit 命令则可以将暂存区中的文件提交…
阅读更多...
在.Net6中用gdal实现第一个功能

在.Net6中用gdal实现第一个功能

目录 一、创建.NET6的控制台应用程序 二、加载Gdal插件 三、编写程序 一、创建.NET6的控制台应用程序 二、加载Gdal插件 Gdal的资源可以经过NuGet包引入。右键单击项目名称&#xff0c;然后选择 "Manage NuGet Packages"&#xff08;管理 NuGet 包&#xff09;。N…
阅读更多...
SD 修复 Midjourney 有瑕疵照片

SD 修复 Midjourney 有瑕疵照片

Midjourney V6 生成的照片在质感上有了一个巨大的提升。下面4张图就是 Midjourney V6 生成的。 如果仔细观察人物和老虎的面部&#xff0c;细节真的很丰富。 但仔细观察上面四张图的手部细节&#xff0c;就会发现至少有两只手是有问题的。这也是目前所有 AI 绘图工具面临的问题…
阅读更多...
阿里云2核4G服务器租用价格30元、165元和199元1年

阿里云2核4G服务器租用价格30元、165元和199元1年

阿里云2核4G服务器租用优惠价格&#xff0c;轻量2核4G服务器165元一年、u1服务器2核4G5M带宽199元一年、云服务器e实例30元3个月&#xff0c;活动链接 aliyunfuwuqi.com/go/aliyun 活动链接如下图&#xff1a; 阿里云2核4G服务器优惠价格 轻量应用服务器2核2G4M带宽、60GB高效…
阅读更多...
thinkadmin 新版安装步骤

thinkadmin 新版安装步骤

1.通过 Composer 安装： ( 推荐方式，默认只安装 admin 模块 ) ### 创建项目（ 需要在英文目录下面执行 ） composer create-project zoujingli/thinkadmin### 进入项目根目录 cd thinkadmin### 数据库初始化并安装 ### 默认使用 Sqlite 数据库，若使用其他数据库请按第二步修…
阅读更多...
大话设计模式之原型模式

大话设计模式之原型模式

原型模式&#xff08;Prototype Pattern&#xff09;是一种创建型设计模式&#xff0c;它用于创建对象的复制&#xff0c;同时又能保持对象的封装。原型模式通过复制现有对象的方式来创建新的对象&#xff0c;而无需知道具体创建过程的细节。 在原型模式中&#xff0c;通常会有…
阅读更多...
经纬恒润AUTOSAR产品成功适配芯来RISC-V车规内核

经纬恒润AUTOSAR产品成功适配芯来RISC-V车规内核

近日&#xff0c;经纬恒润AUTOSAR基础软件产品INTEWORK-EAS&#xff08;ECU AUTOSAR Software&#xff0c;以下简称EAS&#xff09;在芯来提供的HP060开发板上成功适配芯来科技的RISC-V处理器NA内核&#xff0c;双方携手打造了具备灵活、可靠、高性能、强安全性的解决方案。这极…
阅读更多...
C++王牌结构hash：哈希表开散列(哈希桶)的实现与应用

C++王牌结构hash：哈希表开散列(哈希桶)的实现与应用

目录 一、开散列的概念 1.1开散列与闭散列比较 二、开散列/哈希桶的实现 2.1开散列实现 哈希函数的模板构造 哈希表节点构造 开散列增容 插入数据 2.2代码实现 一、开散列的概念 开散列法又叫链地址法(开链法)&#xff0c;首先对关键码集合用散列函数计算散列地址&…
阅读更多...
微软开源项目Garnet：Redis的竞争者还是替代者？

微软开源项目Garnet：Redis的竞争者还是替代者？

对于开源社区&#xff0c;最近的一大新闻就是Redis宣布从7.4版本开始&#xff0c;将采用Redis源代码可用许可证&#xff08;RSALv2&#xff09;和服务器端公共许可证&#xff08;SSPLv1&#xff09;的双重许可证&#xff0c;取代原有的BSD三条款许可证。这一变化引发了开发者社…
阅读更多...
面试算法-126-二叉树的所有路径

面试算法-126-二叉树的所有路径

题目 给你一个二叉树的根节点 root &#xff0c;按 任意顺序 &#xff0c;返回所有从根节点到叶子节点的路径。 叶子节点 是指没有子节点的节点。 示例 1&#xff1a; 输入&#xff1a;root [1,2,3,null,5] 输出&#xff1a;[“1->2->5”,“1->3”] 解 class …
阅读更多...
WIFI驱动移植实验：WIFI从路由器动态获取IP地址与联网

WIFI驱动移植实验：WIFI从路由器动态获取IP地址与联网

一. 简介 前面两篇文章&#xff0c;一篇文章实现了WIFI联网前要做的工作&#xff0c;另一篇文章配置了WIFI配置文件&#xff0c;进行了WIFI热点的连接。文章如下&#xff1a; WIFI驱动移植实验&#xff1a;WIFI 联网前的工作-CSDN博客 WIFI驱动移植实验&#xff1a;连接WIF…
阅读更多...
pdfjs 实现给定pdf数据切片高亮并且跳转

pdfjs 实现给定pdf数据切片高亮并且跳转

pdfjs 实现给定pdf数据切片高亮并且跳转 pdfjs 类的改写基本展示需求的实现高亮功能的实现查询功能分析切片数据处理 pdfjs 类的改写 需求&#xff1a; pdf文件被解析成多个分段&#xff0c;每个分段需要能够展示&#xff0c;并且通过点击分段实现源pdf内容的高亮以及跳转需求…
阅读更多...
204基于matlab的图像融合

204基于matlab的图像融合

基于matlab的图像融合&#xff0c;包括三种方式&#xff0c;加权、PCA、IHS变换。比较三者融合后的图像差异。程序已调通&#xff0c;可直接运行。 204 matlab 图像融合 信息融合 - 小红书 (xiaohongshu.com)
阅读更多...
1.1 单片机的概念

1.1 单片机的概念

一，单片机的概念 单片机（Single-Chip Microcomputer），也被称为单片微控制器，是一种集成电路芯片。它采用超大规模集成电路技术，将具有数据处理能力的中央处理器CPU、随机存储器RAM、只读存储器ROM、多种I/O口和中断系统、定时器/计数器等功能（可能还包括显示驱动电路、…
阅读更多...
Spark SQL— Catalyst 优化器

Spark SQL— Catalyst 优化器

Spark SQL— Catalyst 优化器 1. 目的 本文的目标是描述Spark SQL 优化框架以及它如何允许开发人员用很少的代码行表达复杂的查询转换。我们还将描述Spark SQL如何通过大幅提高其查询优化能力来提高查询的执行时间。在本教程中&#xff0c;我们还将介绍什么是优化、为什么使用…
阅读更多...
《VideoMamba》论文笔记

《VideoMamba》论文笔记

原文链接&#xff1a; [2403.06977] VideoMamba: State Space Model for Efficient Video Understanding (arxiv.org) 原文笔记 What&#xff1a; VideoMamba: State Space Model for Efficient Video Understanding 作者探究Mamba模型能否用于VideoUnderStanding作者引入…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站 广利置业