当前位置: 首页 > news >正文

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来实现延时队列

  1. 首先我们将刚下载下来的rabbitmq_delayed_message_exchange-3.9.0.ez文件上传到RabbitMQ所在服务器,下载地址:https://www.rabbitmq.com/community-plugins.html
  2. 切换到插件所在目录,执行 docker cp rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins 命令,将刚插件拷贝到容器内plugins目录下
  3. 执行 docker exec -it rabbitmq /bin/bash 命令进入到容器内部,并 cd plugins 进入plugins目录
  4. 执行 ls -l|grep delay 命令查看插件是否copy成功
  5. 在容器内plugins目录下,执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 命令启用插件
  6. 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&#xff08;Time To Live&#xff09;1.2 死信交换机 Dead Letter Exchanges1.3 代码实现 2.基于延迟插件实现延迟消息2.1 插件安装2.2 代码实现 3.基于延迟插件封装消息 一、延迟消息 延迟消息有两种实现方案&#xff…...

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有关系的&#xff0c;包括Thread&#xff0c;Looper&#xff0c;Handler&#xff0c;MessageQueue Looper: 由于Looper是android包加入的类&#xff0c;而Thread是java包的类&#xff0c;所以&#xff0c;想要为Thread创建一个Looper&#xff0c;需要在线程内部调用…...

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 中&#xff0c;SET 语句用于更新节点或关系的属性。它允许你修改节点或关系的属性值&#xff0c;可以单独使用&#xff0c;也可以与其他查询语句&#xff08;如 MATCH、CREATE、MERGE 等&#xff09;一起使用。以下是一些使用 SET 语句的常见例子&#xff0c;以及它们…...

Redis 缓存过期及删除

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

万字长文·通俗易懂·一篇包掌握——输入/输出·文件操作(c语言超详细系列)(二)

前言&#xff1a;Hello&#xff0c;大家好&#x1f618;&#xff0c;我是心跳sy&#xff0c;上一节我们主要学习了格式化输入输出的基本内容&#xff0c;这一节我们对格式化进行更加深入的了解&#xff0c;对文件概念进行介绍&#xff0c;并且对输入、输出与文件读写的基本概念…...

【左神算法刷题班】第17节:在有序二维数组中查找目标值、等于目标字符串的子序列个数

第17节 题目1&#xff1a;在有序二维数组中查找目标值 给定一个每一行有序、每一列也有序&#xff0c;整体可能无序的二维数组 再给定一个数num&#xff0c; 返回二维数组中有没有num这个数 例子 数组如下&#xff0c;找 6 是否存在。 1 3 5 7 2 4 6 13 3 9 14 …...

【Terraform学习】本地变量(Terraform配置语言学习)

背景&#xff1a; 关于如何在机器上拉terraform代码&#xff0c;初始化就不重复了&#xff0c;需要的可以查看前面的文章&#xff1a; 【Terraform学习】Terraform-AWS部署快速入门&#xff08;快速入门&#xff09;_向往风的男子的博客-CSDN博客 使用本地变量命名资源 将每…...

zabbix自动注册服务器以及部署代理服务器

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

掌握Python的X篇_32_使用python编辑pdf文件_pdfrw

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

【软件工程】软件测试

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

Android性能优化——内存优化

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

Android Studio实现图形验证码

源代码 源代码MainActivity 效果图32行需要修改&#xff0c;不修改会报错&#xff1a;需要常量表达式&#xff0c;我的代码已修改 点击后 MainActivity import static com.example.graphicverificationcode.RxCaptcha.TYPE.NUMBER;import android.annotation.SuppressLint; …...

JAVA面试数据库篇

目录 一.优化 1.MYSQL中&#xff0c;如何定位慢查询&#xff1f; 2.SQL语句执行慢&#xff0c;如何分析呢&#xff1f; 3.索引 了解过索引吗&#xff1f;&#xff08;什么是索引&#xff09; 索引的底层数据结构了解过吗&#xff1f; B树和B树的区别是什么呢? 什么是聚…...

Android高手进阶教程(三)之----Android 中自定义View的应用.

大家好我们今天的教程是在Android 教程中自定义View 的学习&#xff0c;对于初学着来说&#xff0c;他们习惯了Android 传统的页面布局方式&#xff0c;如下代码: <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"htt…...

第一百一十三回 dart中的getter/setter方法

文章目录 概念介绍使用方法示例代码使用扩展 我们在上一章回中介绍了 flutter_screenutil包相关的内容&#xff0c;本章回中将介绍 dart中的setter/getter方法.闲话休提&#xff0c;让我们一起Talk Flutter吧。 概念介绍 我们在这里介绍的setter/getter方法属于编程语言中的…...

搭建Docker环境

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

微服务08-多级缓存

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

Intel汇编和ATT汇编的区别?

一、前缀不同 在 Intel 语法中&#xff0c;没有寄存器前缀或立即前缀。 然而&#xff0c;在 AT&T 中&#xff0c;寄存器的前缀是“%”&#xff0c;而 immed 的前缀是“$”。 Intel 语法十六进制或二进制即时数据分别带有“h”和“b”后缀。 此外&#xff0c;如果第一个十六…...

ES6从入门到精通:前言

ES6简介 ES6&#xff08;ECMAScript 2015&#xff09;是JavaScript语言的重大更新&#xff0c;引入了许多新特性&#xff0c;包括语法糖、新数据类型、模块化支持等&#xff0c;显著提升了开发效率和代码可维护性。 核心知识点概览 变量声明 let 和 const 取代 var&#xf…...

(十)学生端搭建

本次旨在将之前的已完成的部分功能进行拼装到学生端&#xff0c;同时完善学生端的构建。本次工作主要包括&#xff1a; 1.学生端整体界面布局 2.模拟考场与部分个人画像流程的串联 3.整体学生端逻辑 一、学生端 在主界面可以选择自己的用户角色 选择学生则进入学生登录界面…...

Admin.Net中的消息通信SignalR解释

定义集线器接口 IOnlineUserHub public interface IOnlineUserHub {/// 在线用户列表Task OnlineUserList(OnlineUserList context);/// 强制下线Task ForceOffline(object context);/// 发布站内消息Task PublicNotice(SysNotice context);/// 接收消息Task ReceiveMessage(…...

2025盘古石杯决赛【手机取证】

前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来&#xff0c;实在找不到&#xff0c;希望有大佬教一下我。 还有就会议时间&#xff0c;我感觉不是图片时间&#xff0c;因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...

安宝特方案丨船舶智造的“AR+AI+作业标准化管理解决方案”(装配)

船舶制造装配管理现状&#xff1a;装配工作依赖人工经验&#xff0c;装配工人凭借长期实践积累的操作技巧完成零部件组装。企业通常制定了装配作业指导书&#xff0c;但在实际执行中&#xff0c;工人对指导书的理解和遵循程度参差不齐。 船舶装配过程中的挑战与需求 挑战 (1…...

Linux 中如何提取压缩文件 ?

Linux 是一种流行的开源操作系统&#xff0c;它提供了许多工具来管理、压缩和解压缩文件。压缩文件有助于节省存储空间&#xff0c;使数据传输更快。本指南将向您展示如何在 Linux 中提取不同类型的压缩文件。 1. Unpacking ZIP Files ZIP 文件是非常常见的&#xff0c;要在 …...

Cilium动手实验室: 精通之旅---13.Cilium LoadBalancer IPAM and L2 Service Announcement

Cilium动手实验室: 精通之旅---13.Cilium LoadBalancer IPAM and L2 Service Announcement 1. LAB环境2. L2公告策略2.1 部署Death Star2.2 访问服务2.3 部署L2公告策略2.4 服务宣告 3. 可视化 ARP 流量3.1 部署新服务3.2 准备可视化3.3 再次请求 4. 自动IPAM4.1 IPAM Pool4.2 …...

springboot 日志类切面,接口成功记录日志,失败不记录

springboot 日志类切面&#xff0c;接口成功记录日志&#xff0c;失败不记录 自定义一个注解方法 import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target;/***…...

Spring Security 认证流程——补充

一、认证流程概述 Spring Security 的认证流程基于 过滤器链&#xff08;Filter Chain&#xff09;&#xff0c;核心组件包括 UsernamePasswordAuthenticationFilter、AuthenticationManager、UserDetailsService 等。整个流程可分为以下步骤&#xff1a; 用户提交登录请求拦…...

C++实现分布式网络通信框架RPC(2)——rpc发布端

有了上篇文章的项目的基本知识的了解&#xff0c;现在我们就开始构建项目。 目录 一、构建工程目录 二、本地服务发布成RPC服务 2.1理解RPC发布 2.2实现 三、Mprpc框架的基础类设计 3.1框架的初始化类 MprpcApplication 代码实现 3.2读取配置文件类 MprpcConfig 代码实现…...