6、Spring Boot 3.x集成RabbitMQ动态交换机、队列

一、前言

本篇主要是围绕着 Spring Boot 3.x 与 RabbitMQ 的动态配置集成,比如动态新增 RabbitMQ 交换机、队列等操作。

二、默认RabbitMQ中的exchange、queue动态新增及监听

1、新增RabbitMQ配置

RabbitMQConfig.java
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @className: RabbitConfig* @program: chain* @description: RabbitMQ 配置类* @author: kenny* @create: 2024-10-03 21:59* @version: 1.0.0*/
@Configuration
@EnableRabbit
public class RabbitMQConfig {/*** 创建 RabbitTemplate, 用于发送消息** @return RabbitTemplate*/@Beanpublic RabbitTemplate rabbitTemplate() {return new RabbitTemplate();}/*** 创建 RabbitAdmin, 用于创建 Exchange 和 Queue** @param rabbitTemplate RabbitTemplate* @return RabbitAdmin*/@Beanpublic RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate) {return new RabbitAdmin(rabbitTemplate);}
}

2、新增RabbitMQ动态操作组件

RabbitDynamicConfigService.java
RabbitDynamicConfigService.java 中包含了不同类型Exchange的创建、删除,Queue的创建和删除、绑定Exchange
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.util.Map;/*** @className: RabbitDynamicConfigService* @program: chain* @description: 动态创建队列和交换机* @author: kenny* @create: 2024-10-03 23:49* @version: 1.0.0*/
@Slf4j
@Service
public class RabbitDynamicConfigService {/*** 为了解决循环依赖问题*/private final RabbitAdmin rabbitAdmin;private final RabbitListenerService rabbitListenerService;@Autowiredpublic RabbitDynamicConfigService(RabbitAdmin rabbitAdmin,RabbitListenerService rabbitListenerService) {this.rabbitAdmin = rabbitAdmin;this.rabbitListenerService = rabbitListenerService;}/*** 动态创建队列,并持久化** @param queueName 队列名称*/public void createQueue(String queueName) {// 队列持久化Queue queue = new Queue(queueName, true);// 创建队列rabbitAdmin.declareQueue(queue);System.out.println("队列创建成功: " + queueName);}/*** 动态创建队列,并持久化** @param queueName 队列名称*/public void createQueue(String queueName, Boolean isListener) {// 队列持久化Queue queue = new Queue(queueName, true);// 创建队列rabbitAdmin.declareQueue(queue);System.out.println("队列创建成功: " + queueName);if (!isListener) {return;}rabbitListenerService.createListener(queueName);}/*** 动态创建交换机,并持久化** @param exchangeName 交换机名称*/public void createExchange(String exchangeName) {// 交换机持久化DirectExchange exchange = new DirectExchange(exchangeName, true, false);rabbitAdmin.declareExchange(exchange);log.info("交换机创建成功: {}", exchangeName);}// 动态创建 Fanout 交换机public void createDirectExchange(String exchangeName) {DirectExchange fanoutExchange = new DirectExchange(exchangeName, true, false); // 持久化rabbitAdmin.declareExchange(fanoutExchange);log.info("Direct 交换机创建成功: {}", exchangeName);}// 动态创建 Fanout 交换机public void createFanoutExchange(String exchangeName) {FanoutExchange fanoutExchange = new FanoutExchange(exchangeName, true, false); // 持久化rabbitAdmin.declareExchange(fanoutExchange);log.info("Fanout 交换机创建成功: {}", exchangeName);}// 动态创建 Topic 交换机public void createTopicExchange(String exchangeName) {TopicExchange topicExchange = new TopicExchange(exchangeName, true, false); // 持久化rabbitAdmin.declareExchange(topicExchange);log.info("Topic 交换机创建成功: {}", exchangeName);}// 动态创建 Headers 交换机public void createHeadersExchange(String exchangeName) {HeadersExchange headersExchange = new HeadersExchange(exchangeName, true, false); // 持久化rabbitAdmin.declareExchange(headersExchange);log.info("Headers 交换机创建成功: {}", exchangeName);}/*** 动态绑定队列到交换机,并指定路由键** @param queueName    队列名称* @param exchangeName 交换机名称* @param routingKey   路由键*/public void bindQueueToExchange(String queueName, String exchangeName, String routingKey) {Queue queue = new Queue(queueName);DirectExchange exchange = new DirectExchange(exchangeName);Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);rabbitAdmin.declareBinding(binding);log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);}/*** 动态绑定队列到交换机,并指定路由键** @param queueName    队列名称* @param exchangeName 交换机名称* @param routingKey   路由键*/public void moreExchangeTypeBindQueueToExchange(String queueName, String exchangeType, String exchangeName, String routingKey, Map<String, Object> headers) {switch (exchangeType) {case "fanout" -> bindQueueToExchange(queueName, exchangeName, routingKey);case "direct" -> bindQueueToDirectExchange(queueName, exchangeName, routingKey);case "topic" -> bindQueueToTopicExchange(queueName, exchangeName, routingKey);case "headers" -> bindQueueToHeadersExchange(queueName, exchangeName, headers);default -> throw new IllegalArgumentException("不支持的交换机类型: " + exchangeType);}}/*** 动态绑定队列到交换机,并指定路由键(exchange: direct)** @param queueName    队列名称* @param exchangeName 交换机名称*/public void bindQueueToFanoutExchange(String queueName, String exchangeName) {Queue queue = new Queue(queueName);FanoutExchange exchange = new FanoutExchange(exchangeName);Binding binding = BindingBuilder.bind(queue).to(exchange);rabbitAdmin.declareBinding(binding);log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName);}/*** 动态绑定队列到交换机,并指定路由键(exchange: direct)** @param queueName    队列名称* @param exchangeName 交换机名称* @param routingKey   路由键*/public void bindQueueToDirectExchange(String queueName, String exchangeName, String routingKey) {Queue queue = new Queue(queueName);DirectExchange exchange = new DirectExchange(exchangeName);Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);rabbitAdmin.declareBinding(binding);log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);}/*** 动态绑定队列到交换机,并指定路由键(exchange: topic)** @param queueName    队列名称* @param exchangeName 交换机名称* @param routingKey   路由键*/public void bindQueueToTopicExchange(String queueName, String exchangeName, String routingKey) {Queue queue = new Queue(queueName);TopicExchange exchange = new TopicExchange(exchangeName);Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);rabbitAdmin.declareBinding(binding);log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);}/*** 动态绑定队列到交换机,并指定路由键(exchange: headers)** @param queueName    队列名称* @param exchangeName 交换机名称* @param headers      路由键*/public void bindQueueToHeadersExchange(String queueName, String exchangeName, Map<String, Object> headers) {Queue queue = new Queue(queueName);HeadersExchange exchange = new HeadersExchange(exchangeName);Binding binding = BindingBuilder.bind(queue).to(exchange).whereAll(headers).match();rabbitAdmin.declareBinding(binding);log.info("队列 {}", queueName + " 已绑定到 Headers 交换机 {}", exchangeName + ",使用头部匹配规则: {}", headers);}/*** 动态删除队列** @param queueName 队列名称*/public void deleteQueue(String queueName) {rabbitAdmin.deleteQueue(queueName);log.info("队列删除成功: {}", queueName);}/*** 动态删除交换机** @param exchangeName 交换机名称*/public void deleteExchange(String exchangeName) {rabbitAdmin.deleteExchange(exchangeName);log.info("交换机删除成功: {}", exchangeName);}
}

3、RabbitMQ中队列的动态监听

RabbitListenerService.java
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;/*** @className: RabbitListenerService* @program: chain* @description: RabbitMQ监听器Service组件* @author: kenny* @create: 2024-10-04 01:40* @version: 1.0.0*/
@Slf4j
@Service
public class RabbitListenerService {// 为了解决循环依赖问题private final SimpleRabbitListenerContainerFactory listenerContainerFactory;private final ConnectionFactory connectionFactory;@Autowiredpublic RabbitListenerService(SimpleRabbitListenerContainerFactory listenerContainerFactory,ConnectionFactory connectionFactory) {this.listenerContainerFactory = listenerContainerFactory;this.connectionFactory = connectionFactory;}/*** 创建监听器容器并启动监听** @param queueName 队列名称*/public void createListener(String queueName) {// 创建并启动监听器容器SimpleMessageListenerContainer container = listenerContainerFactory.createListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);// 监听逻辑处理container.setMessageListener(new MessageListenerAdapter(new Object() {public void handleMessage(String message) {System.out.println("收到来自RabbitMQ中队列:" + queueName + " 队列的消息:" + message);}}));// 启动监听器容器container.start();System.out.println("RabbitMQ队列监听器已启动:" + queueName);}
}

4、RabbitMQ中的Exchange、Queue动态操作接口

RabbitDynamicChannelController.java
import com.chain.air.rpp.exchange.rabbit.RabbitDynamicConfigService;
import jakarta.annotation.Resource;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.Map;/*** @className: RabbitDynamicController* @program: chain* @description: RabbitMQ 动态创建队列、交换机,绑定等操作* @author: kenny* @create: 2024-10-04 00:22* @version: 1.0.0*/
@RestController
@RequestMapping("/rabbit/dynamic/channel")
public class RabbitDynamicChannelController {/*** 动态创建队列和交换机*/@Resourceprivate RabbitDynamicConfigService rabbitDynamicConfigService;/*** 动态创建队列** @param queueName 队列名称* @return 处理结果*/@GetMapping("/createQueue")public String createQueue(@RequestParam("queueName") String queueName) {rabbitDynamicConfigService.createQueue(queueName);return "队列已创建: " + queueName;}/*** 动态创建交换机** @param exchangeName 交换机名称* @return 处理结果*/@GetMapping("/createExchange")public String createExchange(@RequestParam("exchangeName") String exchangeName) {rabbitDynamicConfigService.createExchange(exchangeName);return "交换机已创建: " + exchangeName;}/*** 动态绑定队列和交换机** @param queueName    队列名称* @param exchangeName 交换机名称* @param routingKey   路由键* @return 处理结果*/@GetMapping("/bindQueue")public String bindQueueToExchange(@RequestParam("queueName") String queueName,@RequestParam("exchangeName") String exchangeName,@RequestParam("routingKey") String routingKey) {rabbitDynamicConfigService.bindQueueToExchange(queueName, exchangeName, routingKey);return "队列和交换机已绑定: " + queueName + " -> " + exchangeName;}/*** 动态删除队列** @param queueName 队列名称* @return 处理结果*/@GetMapping("/deleteQueue")public String deleteQueue(@RequestParam("queueName") String queueName) {rabbitDynamicConfigService.deleteQueue(queueName);return "队列已删除: " + queueName;}/*** 动态删除交换机** @param exchangeName 交换机名称* @return 处理结果*/@GetMapping("/deleteExchange")public String deleteExchange(@RequestParam("exchangeName") String exchangeName) {rabbitDynamicConfigService.deleteExchange(exchangeName);return "交换机已删除: " + exchangeName;}// 创建并绑定 Fanout 交换机@GetMapping("/createDirectExchange")public String createDirectExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam String routingKey) {rabbitDynamicConfigService.createDirectExchange(exchangeName);rabbitDynamicConfigService.bindQueueToDirectExchange(queueName, exchangeName, routingKey);return "Fanout Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with routing key: " + routingKey;}// 创建并绑定 Fanout 交换机@GetMapping("/createFanoutExchange")public String createFanoutExchange(@RequestParam String exchangeName, @RequestParam String queueName) {rabbitDynamicConfigService.createFanoutExchange(exchangeName);rabbitDynamicConfigService.bindQueueToFanoutExchange(queueName, exchangeName);return "Fanout Exchange and Queue Binding created: " + exchangeName + " -> " + queueName;}// 创建并绑定 Topic 交换机@GetMapping("/createTopicExchange")public String createTopicExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam String routingKey) {rabbitDynamicConfigService.createTopicExchange(exchangeName);rabbitDynamicConfigService.bindQueueToTopicExchange(queueName, exchangeName, routingKey);return "Topic Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with routing key: " + routingKey;}// 创建并绑定 Headers 交换机@GetMapping("/createHeadersExchange")public String createHeadersExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam Map<String, String> headersMap) {Map<String, Object> headers = new HashMap<>(headersMap);rabbitDynamicConfigService.createHeadersExchange(exchangeName);rabbitDynamicConfigService.bindQueueToHeadersExchange(queueName, exchangeName, headers);return "Headers Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with headers: " + headers;}
}

5、RabbitMQ中的Queue消息监听动态操作接口

RabbitChannelListenerController.java
import com.chain.air.rpp.exchange.rabbit.RabbitDynamicConfigService;
import jakarta.annotation.Resource;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;/*** @className: RabbitListenerController* @program: chain* @description: RabbitMQ 监听器 Controller 组件* @author: kenny* @create: 2024-10-04 01:30* @version: 1.0.0*/
@RestController
@RequestMapping("/rabbit/channel/listener")
public class RabbitChannelListenerController {@Resourceprivate RabbitDynamicConfigService rabbitDynamicConfigService;/*** 创建监听器,监听指定队列** @param queueName 队列名称* @return 处理结果*/@GetMapping("/queue")public String listenQueue(@RequestParam("queueName") String queueName) {rabbitDynamicConfigService.createQueue(queueName, true);return "开始监听队列:" + queueName;}
}

三、动态exchange、queue的测试

1、测试Exchange、Queue的动态创建和删除

2、测试Exchange和Queue的动态绑定

3、发送、接收消息测试动态创建Exchange、Queue

4、测试Queue的动态监听接口

下一篇:7、Spring Boot 3.x集成RabbitMQ动态实例等操作

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

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

相关文章

Excel中Ctrl+e的用法

重点&#xff1a;想要使用ctrle&#xff0c;前提是整合或拆分后的结果放置的单元格必须和被提取信息的单元格相邻&#xff0c;且被提取信息的单元格也必须相连。 下图为错误示例 这样则可以使用ctrle 1、信息整合 2、提取信息 3、添加符号 4、信息顺序调换 5、数字提取 crtle还…

HarmonyOS NEXT 应用开发实战(三、ArkUI页面底部导航TabBar的实现)

在开发HarmonyOS NEXT应用时&#xff0c;TabBar是用户界面设计中不可或缺的一部分。本文将通过代码示例&#xff0c;带领大家一同实现一个常用的TabBar&#xff0c;涵盖三个主要的内容页&#xff1a;首页、知乎日报和我的页面。以模仿知乎日报的项目为背景驱动&#xff0c;设定…

解决ubuntu 下 VS code 无法打开点击没反应问题

从Ubuntu 22.04 升级到ubuntu 24.04 后&#xff0c;发现Vsode无法打开&#xff0c;不论是点击图标&#xff0c;还是terminator里面运行code 可执行程序&#xff0c;均没有反应。debug如下: 提示权限不够。 解决方案&#xff1a; sudo sysctl -w kernel.apparmor_restrict_unp…

C语言题目练习2

前面我们知道了单链表的结构及其一些数据操作&#xff0c;今天我们来看看有关于单链表的题目~ 移除链表元素 移除链表元素&#xff1a; https://leetcode.cn/problems/remove-linked-list-elements/description/ 这个题目要求我们删除链表中是指定数据的结点&#xff0c;最终返…

C语言 | Leetcode C语言题解之第460题LFU缓存

题目&#xff1a; 题解&#xff1a; /* 数值链表的节点定义。 */ typedef struct ValueListNode_s {int key;int value;int counter;struct ValueListNode_s *prev;struct ValueListNode_s *next; } ValueListNode;/* 计数链表的节点定义。 其中&#xff0c;head是数值链表的头…

腾讯云Android 与 iOS 相关

移动端&#xff08;Android/iOS&#xff09;支持哪几种系统音量模式&#xff1f; 支持2种系统音量类型&#xff0c;即通话音量类型和媒体音量类型&#xff1a; 通话音量&#xff1a;手机专门为通话场景设计的音量类型&#xff0c;使用手机自带的回声抵消功能&#xff0c;音质…

java使用socket模拟咖啡馆场景,客户与服务器多线程交互场景

java使用socket模拟咖啡馆场景,客户与服务器多线程交互场景 任务的目标是使用客户机 - 服务器架构构建一个“虚拟咖啡馆”系统每个客户端应用程序将扮演 Customer 谁想点茶或咖啡。服务器应用程序将扮演咖啡馆虚拟的角色 Barista 负责处理订单&#xff0c;准备茶和啡&#xff…

OpenCV-风格迁移

文章目录 一、原理二、关键步骤三、实现方法四、可选参数五、示例代码六、总结 OpenCV中的风格迁移是一种计算机视觉技术&#xff0c;它允许用户将一种图像的风格转移到另一幅图像上&#xff0c;从而创造出具有独特美学效果的新图像。这种技术在艺术、设计和娱乐等领域有着广泛…

elementui中table标题和内容插槽

代码 <template> <el-table :data"tableData"> <el-table-column> <template slot"header" slot-scope"scope"> <el-tooltip content"这是列的提示信息" placement"top"> <span>…

谷歌浏览器 文件下载提示网络错误

情况描述&#xff1a; 谷歌版本&#xff1a;129.0.6668.90 (正式版本) &#xff08;64 位&#xff09; (cohort: Control)其他浏览器&#xff0c;比如火狐没有问题&#xff0c;但是谷歌会下载失败&#xff0c;故推断为谷歌浏览器导致的问题小文件比如1、2M会成功&#xff0c;大…

【LeetCode】动态规划—95. 不同的二叉搜索树 II(附完整Python/C++代码)

动态规划—95. 不同的二叉搜索树 II 题目描述前言基本思路1. 问题定义二叉搜索树的性质&#xff1a; 2. 理解问题和递推关系递归构造思想&#xff1a;状态定义&#xff1a;递推公式&#xff1a;终止条件&#xff1a; 3. 解决方法递归 动态规划方法&#xff1a;伪代码&#xff…

sentinel dashboard改造落地设计实现解释(一)-分布式fetcher和metrics存储/搜索

背景 微服务是目前java主流架构,微服务架构技术栈有,服务注册中心,网关,熔断限流,服务同学,配置中心等组件,其中,熔断限流主要3个功能特性,限流,熔断,快速失败。Sentinel是阿里开源的熔断限流组件,sentinel dashboard是演示级别,表现在metrics采集是单机版,metri…

Linux内核 -- 文件系统之超级块 super_operations 字段作用与用法

struct super_operations 字段的作用与用法 super_operations 是 Linux 内核中定义的一个结构体&#xff0c;用于描述与文件系统超级块 (super_block) 相关的操作。这些操作涵盖了文件系统的创建、删除、挂载、卸载等。 结构体定义 struct super_operations {struct inode *…

如何使用vscode的launch.json来debug调试

1、创建一个launch.json文件 选择Python Debugger&#xff0c;再选择Python文件&#xff0c;创建处理如下 默认有下面五个参数 "name": "Python Debugger: Current File","type": "debugpy","request": "launch"…

C++ struct 和 class 的区别、C++结构体和C结构体的区别等基础知识。

特性C structC classC struct默认访问控制符publicprivate全部公开支持成员函数是是否继承支持是是否模板支持是是否运算符重载是是否 在C中&#xff0c;struct 和 class 是两种非常相似的用户定义类型&#xff0c;二者有一些关键区别。C的 struct 与 C 语言中的 struct 也存在…

金九银十软件测试面试题(800道)

今年你的目标是拿下大厂offer&#xff1f;还是多少万年薪&#xff1f;其实这些都离不开日积月累的过程。 为此我特意整理出一份&#xff08;超详细笔记/面试题&#xff09;它几乎涵盖了所有的测试开发技术栈&#xff0c;非常珍贵&#xff0c;人手一份 肝完进大厂 妥妥的&#…

【LeetCode】动态规划—123. 买卖股票的最佳时机 III(附完整Python/C++代码)

动态规划—123. 买卖股票的最佳时机 III 题目描述前言基本思路1. 问题定义2. 理解问题和递推关系状态定义&#xff1a;状态转移公式&#xff1a;初始条件&#xff1a; 3. 解决方法动态规划方法伪代码&#xff1a; 4. 进一步优化5. 小总结 Python代码Python代码解释 C代码C代码解…

07_23 种设计模式之《桥接模式》

文章目录 一、桥接模式基础知识二、桥接模式实战应用 一、桥接模式基础知识 桥接模式定义&#xff1a;将抽象部分与它的实现部分分离&#xff0c;使它们都可以 独立地变化。桥接模式的结构图如下&#xff1a; Abstraction&#xff1a;抽象部分的接口。 RefinedAbstraction&…

Python基础之List列表用法

1、创建列表 names ["张三","李四","王五","Mary"] 2、列表分片 names[1]&#xff1a;获取数组的第2个元素。 names[1:3]&#xff1a;获取数组的第2、第3个元素。包含左侧&#xff0c;不包含右侧。 names[:3]等同于names[0:3]&…

List子接口

1.特点&#xff1a;有序&#xff0c;有下标&#xff0c;元素可以重复 2.方法&#xff1a;包含Collection中的所有方法&#xff0c;还包括自己的独有的方法&#xff08;API中查找&#xff09; 还有ListIterator&#xff08;迭代器&#xff09;&#xff0c;功能更强大。 包含更多…