springBoot对接多个mq并且实现延迟队列---未完待续

mq调用流程

创建消息转换器

package com.wd.config;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqMessageConvertConfig {/*** 公共的消息转换器** @return MessageConverter*/@Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}}

创建exchange交换机:普通交换机、延迟交换机、死信交换机

package com.wd.config;import org.springframework.amqp.core.DirectExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqExchangeConfig {public static final int DELAY_TIME = 20 * 1000;/*** 普通交换机名称*/public static final String EXCHANGE_NAME = "wd_exchange";/*** 延迟交换机名称*/public static final String DELAY_EXCHANGE_NAME = "wd_delay_exchange";/*** 死信交换机*/public static final String DEAD_EXCHANGE_NAME = "wd_dead_exchange";@Beanpublic DirectExchange exchange() {return new DirectExchange(EXCHANGE_NAME, true, false);}@Beanpublic DirectExchange delayExchange() {return new DirectExchange(DELAY_EXCHANGE_NAME, true, false);}@Beanpublic DirectExchange deadExchange() {return new DirectExchange(DEAD_EXCHANGE_NAME, true, false);}
}

创建master的connection

package com.wd.config.master;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqMasterConnectionConfig {@Value("${rabbitmq.master.vhost}")private String vhost;@Value("${rabbitmq.master.addresses}")private String addresses;@Value("${rabbitmq.master.username}")private String username;@Value("${rabbitmq.master.password}")private String password;@Beanpublic ConnectionFactory masterConnectionFactory() {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();cachingConnectionFactory.setAddresses(addresses);cachingConnectionFactory.setVirtualHost(vhost);cachingConnectionFactory.setUsername(username);cachingConnectionFactory.setPassword(password);return cachingConnectionFactory;}@Beanpublic RabbitTemplate rabbitTemplate(@Qualifier("masterConnectionFactory") ConnectionFactory masterConnectionFactory,MessageConverter messageConverter){RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(masterConnectionFactory);rabbitTemplate.setMandatory(true);rabbitTemplate.setMessageConverter(messageConverter);return rabbitTemplate;}}

创建slave的connection

package com.wd.config.slave;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqSlaveConnectionConfig {@Value("${rabbitmq.slave.vhost}")private String vhost;@Value("${rabbitmq.slave.addresses}")private String addresses;@Value("${rabbitmq.slave.username}")private String username;@Value("${rabbitmq.slave.password}")private String password;@Beanpublic ConnectionFactory slaveConnectionFactory() {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();cachingConnectionFactory.setAddresses(addresses);cachingConnectionFactory.setVirtualHost(vhost);cachingConnectionFactory.setUsername(username);cachingConnectionFactory.setPassword(password);return cachingConnectionFactory;}@Beanpublic RabbitTemplate slaveRabbitTemplate(@Qualifier("slaveConnectionFactory") ConnectionFactory slaveConnectionFactory,MessageConverter messageConverter){RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(slaveConnectionFactory);rabbitTemplate.setMandatory(true);rabbitTemplate.setMessageConverter(messageConverter);return rabbitTemplate;}}

创建队列A:  分为普通队列、延迟队列、死信队列

package com.wd.config.queue;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;import static com.wd.config.RabbitmqExchangeConfig.DEAD_EXCHANGE_NAME;
import static com.wd.config.RabbitmqExchangeConfig.DELAY_TIME;@Configuration
public class QueueAConfig {private static final String QUEUE_A_NAME = "wd_queue_a";private static final String DELAY_QUEUE_A_NAME = "wd_delay_queue_a";private static final String DEAD_QUEUE_A_NAME = "wd_dead_queue_a";private static final String QUEUE_A_ROUTING_KEY = "queue_A_routing_key";private static final String DELAY_QUEUE_A_ROUTING_KEY = "delay_queue_a_routing_key";private static final String DEAD_LETTER_QUEUE_A_ROUTING_KEY = "dead_letter_queue_A_routing_key";@Beanpublic Queue queueA() {return new Queue(QUEUE_A_NAME, true);}@Beanpublic Binding queueABinding(@Qualifier("queueA") Queue queueA,@Qualifier("exchange") DirectExchange exchange) {return BindingBuilder.bind(queueA).to(exchange).with(QUEUE_A_ROUTING_KEY);}@Beanpublic Queue delayQueueA() {Map<String, Object> args = new HashMap<>();//设置延迟队列绑定的死信交换机args.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);//设置延迟队列绑定的死信路由键args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_A_ROUTING_KEY);//设置延迟队列的 TTL 消息存活时间args.put("x-message-ttl", DELAY_TIME);return new Queue(DELAY_QUEUE_A_NAME, true, false, false, args);}@Beanpublic Binding delayQueueABinding(@Qualifier("delayQueueA") Queue delayQueueA,@Qualifier("delayExchange") DirectExchange delayExchange) {return BindingBuilder.bind(delayQueueA).to(delayExchange).with(DELAY_QUEUE_A_ROUTING_KEY);}@Beanpublic Queue deadQueueA() {return new Queue(DEAD_QUEUE_A_NAME, true);}@Beanpublic Binding deadQueueABinding(@Qualifier("deadQueueA") Queue deadQueueA,@Qualifier("deadExchange") DirectExchange deadExchange) {return BindingBuilder.bind(deadQueueA).to(deadExchange).with(DEAD_LETTER_QUEUE_A_ROUTING_KEY);}
}

创建队列B: 分为普通队列、延迟队列、死信队列

package com.wd.config.queue;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;import static com.wd.config.RabbitmqExchangeConfig.DEAD_EXCHANGE_NAME;
import static com.wd.config.RabbitmqExchangeConfig.DELAY_TIME;@Configuration
public class QueueBConfig {private static final String QUEUE_B_NAME = "wd_queue_b";private static final String DELAY_QUEUE_B_NAME = "wd_delay_queue_b";private static final String DEAD_QUEUE_B_NAME = "wd_dead_queue_b";private static final String QUEUE_B_ROUTING_KEY = "queue_b_routing_key";private static final String DELAY_QUEUE_B_ROUTING_KEY = "delay_queue_b_routing_key";private static final String DEAD_LETTER_QUEUE_B_ROUTING_KEY = "dead_letter_queue_b_routing_key";@Beanpublic Queue queueB() {return new Queue(QUEUE_B_NAME, true);}@Beanpublic Binding queueBBinding(@Qualifier("queueB") Queue queueB,@Qualifier("exchange") DirectExchange exchange) {return BindingBuilder.bind(queueB).to(exchange).with(QUEUE_B_ROUTING_KEY);}@Beanpublic Queue delayQueueB() {Map<String, Object> args = new HashMap<>();//设置延迟队列绑定的死信交换机args.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);//设置延迟队列绑定的死信路由键args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_B_ROUTING_KEY);//设置延迟队列的 TTL 消息存活时间args.put("x-message-ttl", DELAY_TIME);return new Queue(DELAY_QUEUE_B_NAME, true, false, false, args);}@Beanpublic Binding delayQueueBBinding(@Qualifier("delayQueueB") Queue delayQueueB,@Qualifier("delayExchange") DirectExchange delayExchange) {return BindingBuilder.bind(delayQueueB).to(delayExchange).with(DELAY_QUEUE_B_ROUTING_KEY);}@Beanpublic Queue deadQueueB() {return new Queue(DEAD_QUEUE_B_NAME, true);}@Beanpublic Binding deadQueueABinding(@Qualifier("deadQueueB") Queue deadQueueB,@Qualifier("deadExchange") DirectExchange deadExchange) {return BindingBuilder.bind(deadQueueB).to(deadExchange).with(DEAD_LETTER_QUEUE_B_ROUTING_KEY);}}

创建队列C: 分为普通队列、延迟队列、死信队列

package com.wd.config.queue;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;import static com.wd.config.RabbitmqExchangeConfig.DEAD_EXCHANGE_NAME;
import static com.wd.config.RabbitmqExchangeConfig.DELAY_TIME;@Configuration
public class QueueCConfig {private static final String QUEUE_C_NAME = "wd_queue_c";private static final String DELAY_QUEUE_C_NAME = "wd_delay_queue_c";private static final String DEAD_QUEUE_C_NAME = "wd_dead_queue_c";private static final String QUEUE_C_ROUTING_KEY = "queue_c_routing_key";private static final String DELAY_QUEUE_C_ROUTING_KEY = "delay_queue_c_routing_key";private static final String DEAD_LETTER_QUEUE_C_ROUTING_KEY = "dead_letter_queue_c_routing_key";@Beanpublic Queue queueC() {return new Queue(QUEUE_C_NAME, true);}@Beanpublic Binding queueCBinding(@Qualifier("queueC") Queue queueC,@Qualifier("exchange") DirectExchange exchange) {return BindingBuilder.bind(queueC).to(exchange).with(QUEUE_C_ROUTING_KEY);}@Beanpublic Queue delayQueueC() {Map<String, Object> args = new HashMap<>();//设置延迟队列绑定的死信交换机args.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);//设置延迟队列绑定的死信路由键args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_C_ROUTING_KEY);//设置延迟队列的 TTL 消息存活时间args.put("x-message-ttl", DELAY_TIME);return new Queue(DELAY_QUEUE_C_NAME, true, false, false, args);}@Beanpublic Binding delayQueueCBinding(@Qualifier("delayQueueC") Queue delayQueueC,@Qualifier("delayExchange") DirectExchange delayExchange) {return BindingBuilder.bind(delayQueueC).to(delayExchange).with(DELAY_QUEUE_C_ROUTING_KEY);}@Beanpublic Queue deadQueueC() {return new Queue(DEAD_QUEUE_C_NAME, true);}@Beanpublic Binding deadQueueCBinding(@Qualifier("deadQueueC") Queue deadQueueC,@Qualifier("deadExchange") DirectExchange deadExchange) {return BindingBuilder.bind(deadQueueC).to(deadExchange).with(DEAD_LETTER_QUEUE_C_ROUTING_KEY);}}

创建master的消息监听RabbitListenerContainerFactory

后续使用注解 @RabbitListener 时指定ListenerContainerFactory

@RabbitListener(queues = DEAD_LETTER_QUEUE_B, containerFactory = "masterListenerContainerFactory")
package com.wd.config.master;import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqMasterListenerConfig {@Beanpublic SimpleRabbitListenerContainerFactory masterListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,@Qualifier(value = "masterConnectionFactory") ConnectionFactory masterConnectionFactory,MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置消息转换器factory.setMessageConverter(messageConverter);// 关闭自动ACKfactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);configurer.configure(factory, masterConnectionFactory);return factory;}
}

创建slave的消息监听RabbitListenerContainerFactory

后续使用注解 @RabbitListener 时指定ListenerContainerFactory

@RabbitListener(queues = DEAD_LETTER_QUEUE_B, containerFactory = "slaveListenerContainerFactory")
package com.wd.config.slave;import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitmqSlaveListenerConfig {@Beanpublic SimpleRabbitListenerContainerFactory slaveListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,@Qualifier(value = "slaveConnectionFactory") ConnectionFactory slaveConnectionFactory,MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置消息转换器factory.setMessageConverter(messageConverter);// 关闭自动ACKfactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);configurer.configure(factory, slaveConnectionFactory);return factory;}
}

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

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

相关文章

Vue的详细教程--Vue路由与nodejs

&#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 接下来看看由辉辉所写的关于Vue的相关操作吧 目录 &#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 一.Vue路由是什么 二.使用Vue路由的步骤 1、…

小程序中如何划分会员级别以及不同级别不同积分累计、折扣、返佣、升级条件等

以下是一些关键步骤和注意事项&#xff0c;帮助商家在小程序中有效设置会员卡等级和相关规则。 1. 设定会员卡等级。在管理员后台->会员管理处&#xff0c;点击“等级...”。会出现级别设置界面。 可以创建会员卡等级&#xff0c;并为每个等级设定名称和对应的规则。一般会员…

MyBatis 日志模块

文章目录 前言LogLogFactory日志应用JDBC 日志BaseJdbcLoggerConnectionLogger应用实现 总结 前言 日志在我们开发过程中占据了一个非常重要的地位&#xff0c;是开发和运维管理之间的桥梁&#xff0c;在Java中的日志框架也非常多&#xff0c;Log4j、Log4j2、slf4j等&#xff…

创建表

MySQL从小白到总裁完整教程目录:https://blog.csdn.net/weixin_67859959/article/details/129334507?spm1001.2014.3001.5502 创建表 语法格式: create table 表名(列名1 数据类型,列名2 数据类型,... ,列名n, 数据类型 ); 练习:在czwbkl库中,创建一格test01表 跟大家说…

开发高性能知识付费平台:关键技术策略

引言 在构建知识付费平台时&#xff0c;高性能是确保用户满意度和平台成功的关键因素之一。本文将探讨一些关键的技术策略&#xff0c;帮助开发者打造高性能的知识付费平台。 1. 前端性能优化 使用CDN加速资源加载 使用内容分发网络&#xff08;CDN&#xff09;来托管和加…

Flask数据库之SQLAlchemy--介绍--链接数据库

目录 SQLAlchemy介绍 SQLAlchemy连接数据库 SQLAlchemy介绍 数据库是一个网站的基础&#xff01;&#xff01;&#xff01; 比如MySQL、MongoDB、SQLite、PostgreSQL等&#xff0c;这里我们以MySQL为例进行讲解。 SQLAlchemy是一个ORM框架 对象关系映射&#xff08;英语&…

第一个 Go 程序“hello,world“ 与 main 函数

第一个 Go 程序"hello&#xff0c;world" 与 main 函数 文章目录 第一个 Go 程序"hello&#xff0c;world" 与 main 函数一.创建“hello&#xff0c;world”示例程序二. “hello&#xff0c;world” 程序结构拆解三、main 函数四、Go 语言中程序是怎么编译…

Xpath使用

有如下网页&#xff1a; 需要选中“若出现" 操作如下&#xff1a; 打开Xpath Helper 选中"若出现",右击检查 复制对应的Xpath 在QUERY栏插入即可

[杂谈]-快速了解半波和全波整流

快速了解半波和全波整流 文章目录 快速了解半波和全波整流1、滤波2、半波整流器3、全波整流器4、常见问题 整流器是一种将交流信号转换为脉动直流信号以及将交流电转换为直流电的电子电路。 我们日常生活中几乎所有的电子项目都会用到它。 根据周期传导&#xff0c;本文我们介绍…

VisualBox QA

出现提示注册表错误&#xff0c;或者之前正常&#xff0c;重启VisualBox后&#xff0c;VM运行失败时&#xff0c;可通过正确卸载VisualBox&#xff0c;然后使用注册表清理软件(CCleaner)清理注册表后&#xff0c;重装VisualBox&#xff0c;即会正常。&#xff08;一般用这个能解…

自学网络安全的三个必经阶段(含路线图)

一、为什么选择网络安全&#xff1f; 这几年随着我国《国家网络空间安全战略》《网络安全法》《网络安全等级保护2.0》等一系列政策/法规/标准的持续落地&#xff0c;网络安全行业地位、薪资随之水涨船高。 未来3-5年&#xff0c;是安全行业的黄金发展期&#xff0c;提前踏入…

「UG/NX」BlockUI 选择小平面区域 Select Facet Region

✨博客主页何曾参静谧的博客&#x1f4cc;文章专栏「UG/NX」BlockUI集合&#x1f4da;全部专栏「UG/NX」NX二次开发「UG/NX」BlockUI集合「VS」Visual Studio「QT」QT5程序设计「C/C」C/C程序设计「Win」Windows程序设计「DSA」数据结构与算法「File」数据文件格式 目录 控件说…

vue3+ts 实现移动端分页

current 开始页码 pageSize 结束页码 const sizeref<number>(10) //一页显示十条 const eachCurrentPageref<number>(1) //默认是第一页interface ITdata {current: number,pageSize: number,// xxxx 其他参数... } const selectApplyList ref<…

Verilog 不同编码风格对综合电路的影响

文章目录 示例 #1示例 #2示例 #3 Verilog是一种硬件描述语言&#xff08;HDL&#xff09;&#xff0c;用于设计数字电路和系统。统一、良好的代码编写风格&#xff0c;可以提高代码的可维护性和可读性。 同样的功能&#xff0c;不同的Verilog 编码风格也会对综合过程产生重大影…

1787_函数指针的使用

全部学习汇总&#xff1a;GitHub - GreyZhang/c_basic: little bits of c. 前阵子似乎写了不少错代码&#xff0c;因为对函数指针的理解还不够。今天晚上似乎总算是梳理出了一点眉目&#xff0c;在先前自己写过的代码工程中做一下测试。 先前实现过一个归并排序算法&#xff0c…

【Java毕设项目】基于SpringBoot+Vue校园便利平台的设计与实现

博主主页&#xff1a;一季春秋博主简介&#xff1a;专注Java技术领域和毕业设计项目实战、Java、微信小程序、安卓等技术开发&#xff0c;远程调试部署、代码讲解、文档指导、ppt制作等技术指导。主要内容&#xff1a;毕业设计(Java项目、小程序等)、简历模板、学习资料、面试题…

【LeetCode-中等题】513. 找树左下角的值

文章目录 题目方法一&#xff1a;前序递归方法二&#xff1a;层序遍历 题目 方法一&#xff1a;前序递归 在递归遍历到叶子结点时&#xff0c;对比此时的节点深度&#xff0c;若当前节点深度大于当前最大深度&#xff0c;就更新value值&#xff0c;最后记录下的value即为最下最…

2023.9.23(对这一年过去几个月的总结)

这个时间点杭州正在开亚运会&#xff0c;周六&#xff0c;大周&#xff0c;难得的大周&#xff0c;早上在公司健身房跑完步&#xff0c;就来工位看书了。 反思一下&#xff1a; 技术&#xff1a; 今年在技术学习上的目标&#xff0c;达成率是在太低&#xff0c;但看文章输出来…

手摸手图解 CodeWhisperer 的安装使用

CodeWhisperer 是亚⻢逊出品的一款基于机器学习的通用代码生成器&#xff0c;可实时提供代码建议。 亚马逊云科技开发者社区为开发者们提供全球的开发技术资源。这里有技术文档、开发案例、技术专栏、培训视频、活动与竞赛等。帮助中国开发者对接世界最前沿技术&#xff0c;观点…

【云原生】Kubernetes学习笔记

部署 在部署前强调几点 不要使用IPv6, 很多组件都不支持IPv6不要使用最新版本, 最新版本非常不稳定, 甚至可能存在无法运行的bug不要版本更新, 安装后就将版本固定下来, 新的版本可能会引入新功能, 或移除旧功能, 导致Kubernetes无法运行 Kubeadm介绍 K8s是由多个模块构成的…