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);
}

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

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

相关文章

vue3 setup+Taro3 调用原生小程序自定义年月日时分多列选择器，NutUI改造

vue3 setup+Taro3 调用原生小程序自定义年月日时分多列选择器，NutUI改造

vue3 setupTaro3 调用原生小程序自定义年月日时分多列选择器&#xff0c;NutUI改造 NutUI 有日期时间选择器&#xff0c;但是滑动效果太差&#xff0c;卡顿明显。换成 原生小程序 很顺畅 上代码&#xff1a; <template><view><pickermode"multiSelector&…
阅读更多...
2023牛客暑期多校训练营9-J Puzzle: Star Battle

2023牛客暑期多校训练营9-J Puzzle: Star Battle

2023牛客暑期多校训练营9-J Puzzle: Star Battle https://ac.nowcoder.com/acm/contest/57363/J 文章目录 2023牛客暑期多校训练营9-J Puzzle: Star Battle题意解题思路代码 题意 解题思路 出题人都说是诈骗题&#xff08;&#xff0c;可以发现满足每行每列恰好有 n n n个星…
阅读更多...
【解决】Kafka Exception thrown when sending a message with key=‘null‘ 异常

【解决】Kafka Exception thrown when sending a message with key=‘null‘ 异常

问题原因&#xff1a; 如下图&#xff0c;kafka 中配置的是监听域名的方式&#xff0c;但程序里使用的是 ip:port 的连接方式。 解决办法&#xff1a; kafka 中配置的是域名的方式&#xff0c;程序里也相应配置成 域名:port 的方式&#xff08;注意&#xff1a;本地h…
阅读更多...
机器学习笔记之优化算法(十三)关于二次上界引理

机器学习笔记之优化算法(十三)关于二次上界引理

机器学习笔记之优化算法——关于二次上界引理 引言回顾&#xff1a;利普希兹连续梯度下降法介绍 二次上界引理&#xff1a;介绍与作用二次上界与最优步长之间的关系二次上界引理证明过程 引言 本节将介绍二次上界的具体作用以及它的证明过程。 回顾&#xff1a; 利普希兹连续…
阅读更多...
uniapp 微信小程序 订阅消息

uniapp 微信小程序 订阅消息

第一步&#xff0c;需要先去小程序官方挑选一下订阅模板拿到模板id 订阅按钮在头部导航上&#xff0c;所以 <u-navbar :bgColor"bgColor"><view class"u-nav-slot" slot"left" click"goSubscribe"><image :src"g…
阅读更多...
综述：计算机视觉中的图像分割

综述：计算机视觉中的图像分割

一、说明 这篇文章是关于图像分割的探索&#xff0c;这是解决计算机视觉问题&#xff08;如对象检测、对象识别、图像编辑、医学图像分析、自动驾驶汽车等&#xff09;的重要步骤之一。让我们从介绍开始。 二、图像分割介绍 图像分割是计算机视觉中的一项基本任务&#xff0c;涉…
阅读更多...
【Maven】SpringBoot项目使用maven-assembly-plugin插件多环境打包

【Maven】SpringBoot项目使用maven-assembly-plugin插件多环境打包

SpringBoot项目使用maven-assembly-plugin插件多环境打包 1.创建SpringBoot项目并在pom.xml文件中添加maven-assembly-plugin配置 <!-- 多环境配置 --><profiles><!-- 开发环境 --><profile><id>dev</id><properties><prof…
阅读更多...
新一代分布式融合存储，数据场景All In One

新一代分布式融合存储，数据场景All In One

1、摘要 2023年5月11日&#xff0c;浪潮信息全国巡展广州站正式启航。会上&#xff0c;重磅发布新一代分布式融合存储AS13000G7&#xff0c;其采用极致融合架构设计理念&#xff0c;实现同一套存储满足四种非结构化数据的“All In One”高效融合&#xff0c;数据存力提升300%&a…
阅读更多...
基于WebSocket的在线文字聊天室

基于WebSocket的在线文字聊天室

与Ajax不同&#xff0c;WebSocket可以使服务端主动向客户发送响应&#xff0c;本案例就是基于WebSocket的一个在线聊天室&#xff0c;不过功能比较简单&#xff0c;只能满足文字交流。演示如下。 案例学习于b站up主&#xff0c;链接 。这位up主讲的非常清楚&#xff0c;值得去学…
阅读更多...
item_get_sales-获取TB商品销量详情

item_get_sales-获取TB商品销量详情

一、接口参数说明&#xff1a; item_get_sales-获取商品销量详情&#xff0c;点击更多API调试&#xff0c;请移步注册API账号点击获取测试key和secret 公共参数 请求地址: https://api-gw.onebound.cn/taobao/item_get_sales 名称类型必须描述keyString是调用key&#xff08…
阅读更多...
Idea 快捷键整理

Idea 快捷键整理

Idea快捷键和自动代码补全汇总 idea快捷键汇总 Ctrl 快捷键说明Ctrl F在当前文件进行文本查找 &#xff08;必备&#xff09;Ctrl R在当前文件进行文本替换 &#xff08;必备&#xff09;Ctrl Z撤销 &#xff08;必备&#xff09;Ctrl Y删除光标所在行 或 删除选中的行 &am…
阅读更多...
设计HTML5图像和多媒体

设计HTML5图像和多媒体

在网页中的文本信息直观、明了&#xff0c;而多媒体信息更富内涵和视觉冲击力。恰当使用不同类型的多媒体可以展示个性&#xff0c;突出重点&#xff0c;吸引用户。在HTML5之前&#xff0c;需要借助插件为网页添加多媒体&#xff0c;如Adobe Flash Player、苹果的QuickTime等。…
阅读更多...
【C++精华铺】6.C++类和对象（下）类与对象的知识补充及编译器优化

【C++精华铺】6.C++类和对象（下）类与对象的知识补充及编译器优化

目录 1. 再谈构造 1.1 成员变量的初始化&#xff08;初始化列表&#xff09; 1.2 初始化列表的行为 1.3 explicit关键字 2. 类中的static成员 2.1 静态成员变量 2.2 静态成员函数 3. 友元 3.1 友元函数 3.1 友元类 4. 内部类 5. 匿名对象 6. 对象拷贝时候的编译器优化…
阅读更多...
【LeetCode每日一题】——41.缺失的第一个正数

【LeetCode每日一题】——41.缺失的第一个正数

文章目录 一【题目类别】二【题目难度】三【题目编号】四【题目描述】五【题目示例】六【题目提示】七【解题思路】八【时间频度】九【代码实现】十【提交结果】 一【题目类别】 哈希表 二【题目难度】 困难 三【题目编号】 41.缺失的第一个正数 四【题目描述】 给你一个…
阅读更多...
Compute shader SV 理解图

Compute shader SV 理解图

本图转子&#xff1a;【Computeshader】个人总结_蒋伟博的博客-CSDN博客
阅读更多...
【Rust】Rust学习 第十二章一个 I/O 项目：构建一个命令行程序

【Rust】Rust学习 第十二章一个 I/O 项目：构建一个命令行程序

本章既是一个目前所学的很多技能的概括&#xff0c;也是一个更多标准库功能的探索。我们将构建一个与文件和命令行输入/输出交互的命令行工具来练习现在一些你已经掌握的 Rust 技能。 Rust 的运行速度、安全性、单二进制文件输出和跨平台支持使其成为创建命令行程序的绝佳选择…
阅读更多...
谈一谈在两个商业项目中使用MVI架构后的感悟

谈一谈在两个商业项目中使用MVI架构后的感悟

作者&#xff1a;leobertlan 前言 当时项目采用MVP分层设计&#xff0c;组员的代码风格差异也较大&#xff0c;代码中类职责赋予与封装风格各成一套&#xff0c;随着业务急速膨胀&#xff0c;代码越发混乱。试图用 MVI架构 单向流 形成 掣肘 带来一致风格。 但这种做法不够以…
阅读更多...
yolov5 转换为rknn模型在3588上运行

yolov5 转换为rknn模型在3588上运行

为了把yolov5在rk3588上跑起来&#xff0c;在网上搜罗了一圈,踩了一些坑。由于瑞芯微的文档有升级&#xff0c;导致和网络的文章有出入&#xff0c;所以做个记录。 rknn-toolkit 转换文档&#xff1a; 瑞芯微的转换文档在 rknn-toolkit/example/pytorch/yolov5/REAME.md 里 …
阅读更多...
LangChain入门：构建LLM驱动的应用程序的初学者指南

LangChain入门：构建LLM驱动的应用程序的初学者指南

LangChain & DemoGPT 一、介绍 你有没有想过如何使用大型语言模型&#xff08;LLM&#xff09;构建强大的应用程序&#xff1f;或者&#xff0c;也许您正在寻找一种简化的方式来开发这些应用程序&#xff1f;那么你来对地方了&#xff01;本指南将向您介绍LangChain&#x…
阅读更多...
网络基础--ARP协议介绍

网络基础--ARP协议介绍

1、ARP作用 ARP&#xff08; Address Resolution Protocol&#xff0c;地址解析协议&#xff09;是将 IP 地址解析为以太网 MAC 地址&#xff08;或称物理地址&#xff09;的协议。在局域网中&#xff0c;当主机或其它网络设备有数据要发送给另一个主机或设备时&#xff0c;它必…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023