Rabbitmq延迟消息
目录
- 一、延迟消息
- 1.基于死信实现延迟消息
- 1.1 消息的TTL(Time To Live)
- 1.2 死信交换机 Dead Letter Exchanges
- 1.3 代码实现
- 2.基于延迟插件实现延迟消息
- 2.1 插件安装
- 2.2 代码实现
- 3.基于延迟插件封装消息
一、延迟消息
延迟消息有两种实现方案:
1,基于死信队列
2,集成延迟插件
1.基于死信实现延迟消息
使用RabbitMQ来实现延迟消息必须先了解RabbitMQ的两个概念:
消息的TTL(存活时间)和死信交换机Exchange,通过这两者的组合来实现延迟队列
1.1 消息的TTL(Time To Live)
消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间,也可以对每一个单独的消息做单独的设置。超过了这个时间,我们认为这个消息就死了,称之为死信。
如何设置TTL:
我们创建一个队列queue.temp,在Arguments 中添加x-message-ttl 为5000 (单位是毫秒),那所在压在这个队列的消息在5秒后会消失。
1.2 死信交换机 Dead Letter Exchanges
一个消息在满足如下条件下,会进死信路由,记住这里是路由而不是队列,一个路由可以对应很多队列。
(1) 一个消息被Consumer拒收了,并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里,被其他消费者使用。
(2)上面的消息的TTL到了,消息过期了。
(3)队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。
Dead Letter Exchange其实就是一种普通的exchange,和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了,会自动触发消息的转发,发送到Dead Letter Exchange中去。
我们现在可以测试一下延迟队列。
(1)创建死信队列
(2)创建交换机
(3)建立交换器与队列之间的绑定
(4)创建队列
1.3 代码实现
在service-mq 中添加配置类
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DeadLetterMqConfig {// 声明一些变量public static final String exchange_dead = "exchange.dead";public static final String routing_dead_1 = "routing.dead.1";public static final String routing_dead_2 = "routing.dead.2";public static final String queue_dead_1 = "queue.dead.1";public static final String queue_dead_2 = "queue.dead.2";// 定义交换机@Beanpublic DirectExchange exchange(){return new DirectExchange(exchange_dead,true,false,null);}@Beanpublic Queue queue1(){// 设置如果队列一 出现问题,则通过参数转到exchange_dead,routing_dead_2 上!HashMap<String, Object> map = new HashMap<>();// 参数绑定 此处的key 固定值,不能随意写map.put("x-dead-letter-exchange",exchange_dead);map.put("x-dead-letter-routing-key",routing_dead_2);// 设置延迟时间map.put("x-message-ttl ", 10 * 1000);// 队列名称,是否持久化,是否独享、排外的【true:只可以在本次连接中访问】,是否自动删除,队列的其他属性参数return new Queue(queue_dead_1,true,false,false,map);}@Beanpublic Binding binding(){// 将队列一 通过routing_dead_1 key 绑定到exchange_dead 交换机上return BindingBuilder.bind(queue1()).to(exchange()).with(routing_dead_1);}// 这个队列二就是一个普通队列@Beanpublic Queue queue2(){return new Queue(queue_dead_2,true,false,false,null);}// 设置队列二的绑定规则@Beanpublic Binding binding2(){// 将队列二通过routing_dead_2 key 绑定到exchange_dead交换机上!return BindingBuilder.bind(queue2()).to(exchange()).with(routing_dead_2);}
}
配置发送消息
@RestController
@RequestMapping("/mq")
@Slf4j
public class MqController {@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate RabbitService rabbitService;@GetMapping("sendDeadLettle")public Result sendDeadLettle() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");this.rabbitTemplate.convertAndSend(DeadLetterMqConfig.exchange_dead, DeadLetterMqConfig.routing_dead_1, "ok");System.out.println(sdf.format(new Date()) + " Delay sent.");return Result.ok();}
}
消息接收方
@Component
public class DeadLetterReceiver {@RabbitListener(queues = DeadLetterMqConfig.queue_dead_2)public void getMessage(String msg, Message message, Channel channel) throws IOException {//时间格式化SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("消息接收的时间:\t"+simpleDateFormat.format(new Date()));System.out.println("消息的内容"+msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(),true);}
}
2.基于延迟插件实现延迟消息
2.1 插件安装
Rabbitmq实现了一个插件x-delay-message来实现延时队列
- 首先我们将刚下载下来的rabbitmq_delayed_message_exchange-3.9.0.ez文件上传到RabbitMQ所在服务器,下载地址:https://www.rabbitmq.com/community-plugins.html
- 切换到插件所在目录,执行 docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins 命令,将刚插件拷贝到容器内plugins目录下
- 执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部,并 cd plugins 进入plugins目录
- 执行 ls -l|grep delay 命令查看插件是否copy成功
- 在容器内plugins目录下,执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令启用插件
- exit命令退出RabbitMQ容器内部,然后执行 docker restart rabbitmq 命令重启RabbitMQ容器
2.2 代码实现
配置队列
@Configuration
public class DelayedMqConfig {public static final String exchange_delay = "exchange.delay";public static final String routing_delay = "routing.delay";public static final String queue_delay_1 = "queue.delay.1";@Beanpublic Queue delayQeue1() {// 第一个参数是创建的queue的名字,第二个参数是是否支持持久化return new Queue(queue_delay_1, true);}@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<String, Object>();args.put("x-delayed-type", "direct");return new CustomExchange(exchange_delay, "x-delayed-message", true, false, args);}@Beanpublic Binding delayBbinding1() {return BindingBuilder.bind(delayQeue1()).to(delayExchange()).with(routing_delay).noargs();}
}
发送消息
@GetMapping("sendelay")
public Result sendDelay() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");this.rabbitTemplate.convertAndSend(DelayedMqConfig.exchange_delay, DelayedMqConfig.routing_delay, sdf.format(new Date()), new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDelay(10 * 1000);System.out.println(sdf.format(new Date()) + " Delay sent.");return message;}});return Result.ok();
}
接收消息
@Component
public class DelayReceiver {@RabbitListener(queues = DelayedMqConfig.queue_delay_1)public void get(String msg) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("Receive queue_delay_1: " + sdf.format(new Date()) + " Delay rece." + msg);}}
3.基于延迟插件封装消息
/*** 封装发送延迟消息方法* @param exchange* @param routingKey* @param msg* @param delayTime* @return*/
public Boolean sendDelayMsg(String exchange,String routingKey, Object msg, int delayTime){// 将发送的消息 赋值到 自定义的实体类GmallCorrelationData gmallCorrelationData = new GmallCorrelationData();// 声明一个correlationId的变量String correlationId = UUID.randomUUID().toString().replaceAll("-","");gmallCorrelationData.setId(correlationId);gmallCorrelationData.setExchange(exchange);gmallCorrelationData.setRoutingKey(routingKey);gmallCorrelationData.setMessage(msg);gmallCorrelationData.setDelayTime(delayTime);gmallCorrelationData.setDelay(true);// 将数据存到缓存this.redisTemplate.opsForValue().set(correlationId,JSON.toJSONString(gmallCorrelationData),10,TimeUnit.MINUTES);// 发送消息this.rabbitTemplate.convertAndSend(exchange,routingKey,msg,message -> {// 设置延迟时间message.getMessageProperties().setDelay(delayTime*1000);return message;},gmallCorrelationData);// 默认返回return true;
}
修改retrySendMsg方法 – 添加判断是否属于延迟消息
// 判断是否属于延迟消息
if (gmallCorrelationData.isDelay()){// 属于延迟消息this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),message -> {// 设置延迟时间message.getMessageProperties().setDelay(gmallCorrelationData.getDelayTime()*1000);return message;},gmallCorrelationData);
}else {// 调用发送消息方法 表示发送普通消息 发送消息的时候,不能调用 new RabbitService().sendMsg() 这个方法this.rabbitTemplate.convertAndSend(gmallCorrelationData.getExchange(),gmallCorrelationData.getRoutingKey(),gmallCorrelationData.getMessage(),gmallCorrelationData);
}
利用封装好的工具类 测试发送延迟消息
// 基于延迟插件的延迟消息
@GetMapping("sendDelay")
public Result sendDelay(){// 声明一个时间对象SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("发送时间:"+simpleDateFormat.format(new Date()));this.rabbitService.sendDelayMsg(DelayedMqConfig.exchange_delay,DelayedMqConfig.routing_delay,"iuok",3);return Result.ok();
}
重试了4次,所以我们需要保证幂等性
结果会 回发送三次,也被消费三次!
如何保证消息幂等性?
1.使用数据方式
2.使用redis setnx 命令解决 — 推荐
@SneakyThrows
@RabbitListener(queues = DelayedMqConfig.queue_delay_1)
public void getMsg2(String msg,Message message,Channel channel){// 使用setnx 命令来解决 msgKey = delay:iuokString msgKey = "delay:"+msg;Boolean result = this.redisTemplate.opsForValue().setIfAbsent(msgKey, "0", 10, TimeUnit.MINUTES);// result = true : 说明执行成功,redis 里面没有这个key ,第一次创建, 第一次消费。// result = false : 说明执行失败,redis 里面有这个key// 不能: 那么就表示这个消息只能被消费一次! 那么第一次消费成功或失败,我们确定不了! --- 只能被消费一次!// if (result){// SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// System.out.println("接收时间:"+simpleDateFormat.format(new Date()));// System.out.println("接收的消息:"+msg);// // 手动确认消息// channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);// } else {// // 不能消费!// }// 能: 保证消息被消费成功 第二次消费,可以进来,但是要判断上一个消费者,是否将消息消费了。如果消费了,则直接返回,如果没有消费成功,我消费。// 在设置key 的时候给了一个默认值 0 ,如果消费成功,则将key的值 改为1if (!result){// 获取缓存key对应的数据String status = (String) this.redisTemplate.opsForValue().get(msgKey);if ("1".equals(status)){// 手动确认channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;} else {// 说明第一个消费者没有消费成功,所以消费并确认SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("接收时间:"+simpleDateFormat.format(new Date()));System.out.println("接收的消息:"+msg);// 修改redis 中的数据this.redisTemplate.opsForValue().set(msgKey,"1");channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;}}SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("接收时间:"+simpleDateFormat.format(new Date()));System.out.println("接收的消息:"+msg);// 修改redis 中的数据this.redisTemplate.opsForValue().set(msgKey,"1");// 手动确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}
相关文章:

Rabbitmq延迟消息
目录 一、延迟消息1.基于死信实现延迟消息1.1 消息的TTL(Time To Live)1.2 死信交换机 Dead Letter Exchanges1.3 代码实现 2.基于延迟插件实现延迟消息2.1 插件安装2.2 代码实现 3.基于延迟插件封装消息 一、延迟消息 延迟消息有两种实现方案ÿ…...

miniExcel 生成excel
一、nuget dotnet add package MiniExcel --version 1.31.2 二、新建表及数据 ExampleProducts 三、这里我用了Dapper.Query方法 读取excel public virtual async Task<IActionResult> Anonymous(){try{//using (var connection _dbContext.GetDbConnection())//{//…...

Handler详解
跟Handler有关系的,包括Thread,Looper,Handler,MessageQueue Looper: 由于Looper是android包加入的类,而Thread是java包的类,所以,想要为Thread创建一个Looper,需要在线程内部调用…...
Feign忽略Https的SSL最佳方案(且保证负载均衡将失效)
同时解决Https的SSL证书验证问题和feign不支持Patch请求方法的问题 代码 1. 工具类 OkHttpUtils.java import javax.net.ssl.*; import java.security.KeyManagementException; import java.security.NoSuchAlgorithmException; import java.security.SecureRandom; import j…...
Neo4j之SET基础
在 Neo4j 中,SET 语句用于更新节点或关系的属性。它允许你修改节点或关系的属性值,可以单独使用,也可以与其他查询语句(如 MATCH、CREATE、MERGE 等)一起使用。以下是一些使用 SET 语句的常见例子,以及它们…...

Redis 缓存过期及删除
一、Redis缓存过期策略 物理内存达到上限后,像磁盘空间申请虚拟内存(硬盘与内存的swap),甚至崩溃。 内存与硬盘交换 (swap) 虚拟内存,频繁I0 性能急剧下降,会造成redis内存急剧下降; 一般设置物理内存的3/4,在redis…...

万字长文·通俗易懂·一篇包掌握——输入/输出·文件操作(c语言超详细系列)(二)
前言:Hello,大家好😘,我是心跳sy,上一节我们主要学习了格式化输入输出的基本内容,这一节我们对格式化进行更加深入的了解,对文件概念进行介绍,并且对输入、输出与文件读写的基本概念…...

【左神算法刷题班】第17节:在有序二维数组中查找目标值、等于目标字符串的子序列个数
第17节 题目1:在有序二维数组中查找目标值 给定一个每一行有序、每一列也有序,整体可能无序的二维数组 再给定一个数num, 返回二维数组中有没有num这个数 例子 数组如下,找 6 是否存在。 1 3 5 7 2 4 6 13 3 9 14 …...
【Terraform学习】本地变量(Terraform配置语言学习)
背景: 关于如何在机器上拉terraform代码,初始化就不重复了,需要的可以查看前面的文章: 【Terraform学习】Terraform-AWS部署快速入门(快速入门)_向往风的男子的博客-CSDN博客 使用本地变量命名资源 将每…...

zabbix自动注册服务器以及部署代理服务器
文章目录 Zabbix自动注册服务器及部署代理服务器一.zabbix自动注册1.什么是自动注册2.环境准备3.zabbix客户端配置4.在 Web 页面配置自动注册5.验证自动注册 二.部署 zabbix 代理服务器1.分布式监控的作用:2.环境部署3.代理服务器配置4.客户端配置5.web页面配置5.1 …...

掌握Python的X篇_32_使用python编辑pdf文件_pdfrw
本篇介绍利用python操作pdf文件,我们平时也会有合并和拆分pdf的需求,此时我们就可以使用本节内容。 文章目录 1. pdfrw的安装2. 切分pdf文件3. pdfrw官网及实现一版四面的实例 1. pdfrw的安装 pip install pdfrw官网地址:https://github.co…...

【软件工程】软件测试
软件测试的对象 软件程序文档 测试对象:各个阶段产生的源程序和文档。 软件测试的目的 基于不同的立场,对软件测试的目的存在着两种完全对立的观点。 (1)一种观点是通过测试暴露出软件中所包含的故障和缺陷(从用户的角度)…...

Android性能优化——内存优化
一、内存问题 内存抖动,锯齿状,GC导致卡顿内存泄漏,可用内存减少,频繁GC 内存溢出,OOM,程序异常 二、内存分析工具 Memory ProfilerMemory Analyzer LeakCanary Memory Profiler 实时图表展示应用内存使…...

Android Studio实现图形验证码
源代码 源代码MainActivity 效果图32行需要修改,不修改会报错:需要常量表达式,我的代码已修改 点击后 MainActivity import static com.example.graphicverificationcode.RxCaptcha.TYPE.NUMBER;import android.annotation.SuppressLint; …...
JAVA面试数据库篇
目录 一.优化 1.MYSQL中,如何定位慢查询? 2.SQL语句执行慢,如何分析呢? 3.索引 了解过索引吗?(什么是索引) 索引的底层数据结构了解过吗? B树和B树的区别是什么呢? 什么是聚…...

Android高手进阶教程(三)之----Android 中自定义View的应用.
大家好我们今天的教程是在Android 教程中自定义View 的学习,对于初学着来说,他们习惯了Android 传统的页面布局方式,如下代码: <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"htt…...
第一百一十三回 dart中的getter/setter方法
文章目录 概念介绍使用方法示例代码使用扩展 我们在上一章回中介绍了 flutter_screenutil包相关的内容,本章回中将介绍 dart中的setter/getter方法.闲话休提,让我们一起Talk Flutter吧。 概念介绍 我们在这里介绍的setter/getter方法属于编程语言中的…...

搭建Docker环境
目录 一、docker环境搭建 1、卸载旧版本docker 2、安装依赖和设置仓库 3、安装docker 4、启动并加入开机启动 5、验证是否安装成功 二、利用docker搭建nginx 1、拉取镜像 2、启动容器,部署nginx 一、docker环境搭建 1、卸载旧版本docker yum remove docke…...

微服务08-多级缓存
1.什么是多级缓存 传统的缓存策略一般是请求到达Tomcat后,先查询Redis,如果未命中则查询数据库,如图: 存在下面的问题: •请求要经过Tomcat处理,Tomcat的性能成为整个系统的瓶颈 •Redis缓存失效时,会对数据库产生冲击 多级缓存就是充分利用请求处理的每个环节,分…...

Intel汇编和ATT汇编的区别?
一、前缀不同 在 Intel 语法中,没有寄存器前缀或立即前缀。 然而,在 AT&T 中,寄存器的前缀是“%”,而 immed 的前缀是“$”。 Intel 语法十六进制或二进制即时数据分别带有“h”和“b”后缀。 此外,如果第一个十六…...
设计模式和设计原则回顾
设计模式和设计原则回顾 23种设计模式是设计原则的完美体现,设计原则设计原则是设计模式的理论基石, 设计模式 在经典的设计模式分类中(如《设计模式:可复用面向对象软件的基础》一书中),总共有23种设计模式,分为三大类: 一、创建型模式(5种) 1. 单例模式(Sing…...

现代密码学 | 椭圆曲线密码学—附py代码
Elliptic Curve Cryptography 椭圆曲线密码学(ECC)是一种基于有限域上椭圆曲线数学特性的公钥加密技术。其核心原理涉及椭圆曲线的代数性质、离散对数问题以及有限域上的运算。 椭圆曲线密码学是多种数字签名算法的基础,例如椭圆曲线数字签…...

C++ Visual Studio 2017厂商给的源码没有.sln文件 易兆微芯片下载工具加开机动画下载。
1.先用Visual Studio 2017打开Yichip YC31xx loader.vcxproj,再用Visual Studio 2022打开。再保侟就有.sln文件了。 易兆微芯片下载工具加开机动画下载 ExtraDownloadFile1Info.\logo.bin|0|0|10D2000|0 MFC应用兼容CMD 在BOOL CYichipYC31xxloaderDlg::OnIni…...
在web-view 加载的本地及远程HTML中调用uniapp的API及网页和vue页面是如何通讯的?
uni-app 中 Web-view 与 Vue 页面的通讯机制详解 一、Web-view 简介 Web-view 是 uni-app 提供的一个重要组件,用于在原生应用中加载 HTML 页面: 支持加载本地 HTML 文件支持加载远程 HTML 页面实现 Web 与原生的双向通讯可用于嵌入第三方网页或 H5 应…...
Python 包管理器 uv 介绍
Python 包管理器 uv 全面介绍 uv 是由 Astral(热门工具 Ruff 的开发者)推出的下一代高性能 Python 包管理器和构建工具,用 Rust 编写。它旨在解决传统工具(如 pip、virtualenv、pip-tools)的性能瓶颈,同时…...

R语言速释制剂QBD解决方案之三
本文是《Quality by Design for ANDAs: An Example for Immediate-Release Dosage Forms》第一个处方的R语言解决方案。 第一个处方研究评估原料药粒径分布、MCC/Lactose比例、崩解剂用量对制剂CQAs的影响。 第二处方研究用于理解颗粒外加硬脂酸镁和滑石粉对片剂质量和可生产…...
虚拟电厂发展三大趋势:市场化、技术主导、车网互联
市场化:从政策驱动到多元盈利 政策全面赋能 2025年4月,国家发改委、能源局发布《关于加快推进虚拟电厂发展的指导意见》,首次明确虚拟电厂为“独立市场主体”,提出硬性目标:2027年全国调节能力≥2000万千瓦࿰…...

Git 3天2K星标:Datawhale 的 Happy-LLM 项目介绍(附教程)
引言 在人工智能飞速发展的今天,大语言模型(Large Language Models, LLMs)已成为技术领域的焦点。从智能写作到代码生成,LLM 的应用场景不断扩展,深刻改变了我们的工作和生活方式。然而,理解这些模型的内部…...

数学建模-滑翔伞伞翼面积的设计,运动状态计算和优化 !
我们考虑滑翔伞的伞翼面积设计问题以及运动状态描述。滑翔伞的性能主要取决于伞翼面积、气动特性以及飞行员的重量。我们的目标是建立数学模型来描述滑翔伞的运动状态,并优化伞翼面积的设计。 一、问题分析 滑翔伞在飞行过程中受到重力、升力和阻力的作用。升力和阻力与伞翼面…...
Python 高效图像帧提取与视频编码:实战指南
Python 高效图像帧提取与视频编码:实战指南 在音视频处理领域,图像帧提取与视频编码是基础但极具挑战性的任务。Python 结合强大的第三方库(如 OpenCV、FFmpeg、PyAV),可以高效处理视频流,实现快速帧提取、压缩编码等关键功能。本文将深入介绍如何优化这些流程,提高处理…...