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

盛世宏图  金牌会员 | 2024-10-13 09:15:59 | 来自手机 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 885|帖子 885|积分 2655

一、前言

  1. 本篇主要是围绕着 Spring Boot 3.x 与 RabbitMQ 的动态配置集成，比如动态新增 RabbitMQ 交换机、队列等操作。
复制代码
二、默认RabbitMQ中的exchange、queue动态新增及监听

1、新增RabbitMQ设置

RabbitMQConfig.java

  1. import org.springframework.amqp.rabbit.annotation.EnableRabbit;
  2. import org.springframework.amqp.rabbit.core.RabbitAdmin;
  3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  7. /**
  8. * @className: RabbitConfig
  9. * @program: chain
  10. * @description: RabbitMQ 配置类
  11. * @author: kenny
  12. * @create: 2024-10-03 21:59
  13. * @version: 1.0.0
  14. */
  15. @Configuration
  16. @EnableRabbit
  17. public class RabbitMQConfig {
  19.     /**
  20.      * 创建 RabbitTemplate, 用于发送消息
  21.      *
  22.      * @return RabbitTemplate
  23.      */
  24.     @Bean
  25.     public RabbitTemplate rabbitTemplate() {
  26.         return new RabbitTemplate();
  27.     }
  29.     /**
  30.      * 创建 RabbitAdmin, 用于创建 Exchange 和 Queue
  31.      *
  32.      * @param rabbitTemplate RabbitTemplate
  33.      * @return RabbitAdmin
  34.      */
  35.     @Bean
  36.     public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate) {
  37.         return new RabbitAdmin(rabbitTemplate);
  38.     }
  39. }
复制代码
2、新增RabbitMQ动态操作组件

RabbitDynamicConfigService.java

  1. RabbitDynamicConfigService.java 中包含了不同类型Exchange的创建、删除，Queue的创建和删除、绑定Exchange
复制代码
  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.amqp.core.*;
  3. import org.springframework.amqp.rabbit.core.RabbitAdmin;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Service;
  7. import java.util.Map;
  9. /**
  10. * @className: RabbitDynamicConfigService
  11. * @program: chain
  12. * @description: 动态创建队列和交换机
  13. * @author: kenny
  14. * @create: 2024-10-03 23:49
  15. * @version: 1.0.0
  16. */
  17. @Slf4j
  18. @Service
  19. public class RabbitDynamicConfigService {
  21.     /**
  22.      * 为了解决循环依赖问题
  23.      */
  24.     private final RabbitAdmin rabbitAdmin;
  25.     private final RabbitListenerService rabbitListenerService;
  27.     @Autowired
  28.     public RabbitDynamicConfigService(RabbitAdmin rabbitAdmin,
  29.                                       RabbitListenerService rabbitListenerService) {
  30.         this.rabbitAdmin = rabbitAdmin;
  31.         this.rabbitListenerService = rabbitListenerService;
  32.     }
  34.     /**
  35.      * 动态创建队列，并持久化
  36.      *
  37.      * @param queueName 队列名称
  38.      */
  39.     public void createQueue(String queueName) {
  40.         // 队列持久化
  41.         Queue queue = new Queue(queueName, true);
  42.         // 创建队列
  43.         rabbitAdmin.declareQueue(queue);
  44.         System.out.println("队列创建成功: " + queueName);
  45.     }
  47.     /**
  48.      * 动态创建队列，并持久化
  49.      *
  50.      * @param queueName 队列名称
  51.      */
  52.     public void createQueue(String queueName, Boolean isListener) {
  53.         // 队列持久化
  54.         Queue queue = new Queue(queueName, true);
  55.         // 创建队列
  56.         rabbitAdmin.declareQueue(queue);
  57.         System.out.println("队列创建成功: " + queueName);
  59.         if (!isListener) {
  60.             return;
  61.         }
  63.         rabbitListenerService.createListener(queueName);
  64.     }
  67.     /**
  68.      * 动态创建交换机，并持久化
  69.      *
  70.      * @param exchangeName 交换机名称
  71.      */
  72.     public void createExchange(String exchangeName) {
  73.         // 交换机持久化
  74.         DirectExchange exchange = new DirectExchange(exchangeName, true, false);
  75.         rabbitAdmin.declareExchange(exchange);
  76.         log.info("交换机创建成功: {}", exchangeName);
  77.     }
  80.     // 动态创建 Fanout 交换机
  81.     public void createDirectExchange(String exchangeName) {
  82.         DirectExchange fanoutExchange = new DirectExchange(exchangeName, true, false); // 持久化
  83.         rabbitAdmin.declareExchange(fanoutExchange);
  84.         log.info("Direct 交换机创建成功: {}", exchangeName);
  85.     }
  87.     // 动态创建 Fanout 交换机
  88.     public void createFanoutExchange(String exchangeName) {
  89.         FanoutExchange fanoutExchange = new FanoutExchange(exchangeName, true, false); // 持久化
  90.         rabbitAdmin.declareExchange(fanoutExchange);
  91.         log.info("Fanout 交换机创建成功: {}", exchangeName);
  92.     }
  94.     // 动态创建 Topic 交换机
  95.     public void createTopicExchange(String exchangeName) {
  96.         TopicExchange topicExchange = new TopicExchange(exchangeName, true, false); // 持久化
  97.         rabbitAdmin.declareExchange(topicExchange);
  98.         log.info("Topic 交换机创建成功: {}", exchangeName);
  99.     }
  101.     // 动态创建 Headers 交换机
  102.     public void createHeadersExchange(String exchangeName) {
  103.         HeadersExchange headersExchange = new HeadersExchange(exchangeName, true, false); // 持久化
  104.         rabbitAdmin.declareExchange(headersExchange);
  105.         log.info("Headers 交换机创建成功: {}", exchangeName);
  106.     }
  108.     /**
  109.      * 动态绑定队列到交换机，并指定路由键
  110.      *
  111.      * @param queueName    队列名称
  112.      * @param exchangeName 交换机名称
  113.      * @param routingKey   路由键
  114.      */
  115.     public void bindQueueToExchange(String queueName, String exchangeName, String routingKey) {
  116.         Queue queue = new Queue(queueName);
  117.         DirectExchange exchange = new DirectExchange(exchangeName);
  118.         Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);
  119.         rabbitAdmin.declareBinding(binding);
  120.         log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);
  121.     }
  123.     /**
  124.      * 动态绑定队列到交换机，并指定路由键
  125.      *
  126.      * @param queueName    队列名称
  127.      * @param exchangeName 交换机名称
  128.      * @param routingKey   路由键
  129.      */
  130.     public void moreExchangeTypeBindQueueToExchange(String queueName, String exchangeType, String exchangeName, String routingKey, Map<String, Object> headers) {
  131.         switch (exchangeType) {
  132.             case "fanout" -> bindQueueToExchange(queueName, exchangeName, routingKey);
  133.             case "direct" -> bindQueueToDirectExchange(queueName, exchangeName, routingKey);
  134.             case "topic" -> bindQueueToTopicExchange(queueName, exchangeName, routingKey);
  135.             case "headers" -> bindQueueToHeadersExchange(queueName, exchangeName, headers);
  136.             default -> throw new IllegalArgumentException("不支持的交换机类型: " + exchangeType);
  137.         }
  138.     }
  140.     /**
  141.      * 动态绑定队列到交换机，并指定路由键（exchange: direct）
  142.      *
  143.      * @param queueName    队列名称
  144.      * @param exchangeName 交换机名称
  145.      */
  146.     public void bindQueueToFanoutExchange(String queueName, String exchangeName) {
  147.         Queue queue = new Queue(queueName);
  148.         FanoutExchange exchange = new FanoutExchange(exchangeName);
  149.         Binding binding = BindingBuilder.bind(queue).to(exchange);
  150.         rabbitAdmin.declareBinding(binding);
  151.         log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName);
  152.     }
  154.     /**
  155.      * 动态绑定队列到交换机，并指定路由键（exchange: direct）
  156.      *
  157.      * @param queueName    队列名称
  158.      * @param exchangeName 交换机名称
  159.      * @param routingKey   路由键
  160.      */
  161.     public void bindQueueToDirectExchange(String queueName, String exchangeName, String routingKey) {
  162.         Queue queue = new Queue(queueName);
  163.         DirectExchange exchange = new DirectExchange(exchangeName);
  164.         Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);
  165.         rabbitAdmin.declareBinding(binding);
  166.         log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);
  167.     }
  169.     /**
  170.      * 动态绑定队列到交换机，并指定路由键（exchange: topic）
  171.      *
  172.      * @param queueName    队列名称
  173.      * @param exchangeName 交换机名称
  174.      * @param routingKey   路由键
  175.      */
  176.     public void bindQueueToTopicExchange(String queueName, String exchangeName, String routingKey) {
  177.         Queue queue = new Queue(queueName);
  178.         TopicExchange exchange = new TopicExchange(exchangeName);
  179.         Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey);
  180.         rabbitAdmin.declareBinding(binding);
  181.         log.info("绑定创建成功: {}", queueName + " -> {}", exchangeName + " 使用路由键: {}", routingKey);
  182.     }
  184.     /**
  185.      * 动态绑定队列到交换机，并指定路由键（exchange: headers）
  186.      *
  187.      * @param queueName    队列名称
  188.      * @param exchangeName 交换机名称
  189.      * @param headers      路由键
  190.      */
  191.     public void bindQueueToHeadersExchange(String queueName, String exchangeName, Map<String, Object> headers) {
  192.         Queue queue = new Queue(queueName);
  193.         HeadersExchange exchange = new HeadersExchange(exchangeName);
  194.         Binding binding = BindingBuilder.bind(queue).to(exchange).whereAll(headers).match();
  195.         rabbitAdmin.declareBinding(binding);
  196.         log.info("队列 {}", queueName + " 已绑定到 Headers 交换机 {}", exchangeName + "，使用头部匹配规则: {}", headers);
  197.     }
  199.     /**
  200.      * 动态删除队列
  201.      *
  202.      * @param queueName 队列名称
  203.      */
  204.     public void deleteQueue(String queueName) {
  205.         rabbitAdmin.deleteQueue(queueName);
  206.         log.info("队列删除成功: {}", queueName);
  207.     }
  209.     /**
  210.      * 动态删除交换机
  211.      *
  212.      * @param exchangeName 交换机名称
  213.      */
  214.     public void deleteExchange(String exchangeName) {
  215.         rabbitAdmin.deleteExchange(exchangeName);
  216.         log.info("交换机删除成功: {}", exchangeName);
  217.     }
  218. }
复制代码
3、RabbitMQ中队列的动态监听

RabbitListenerService.java

  1. import lombok.extern.slf4j.Slf4j;
  2. import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
  3. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  4. import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
  5. import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Service;
  9. /**
  10. * @className: RabbitListenerService
  11. * @program: chain
  12. * @description: RabbitMQ监听器Service组件
  13. * @author: kenny
  14. * @create: 2024-10-04 01:40
  15. * @version: 1.0.0
  16. */
  17. @Slf4j
  18. @Service
  19. public class RabbitListenerService {
  21.     // 为了解决循环依赖问题
  22.     private final SimpleRabbitListenerContainerFactory listenerContainerFactory;
  23.     private final ConnectionFactory connectionFactory;
  25.     @Autowired
  26.     public RabbitListenerService(
  27.             SimpleRabbitListenerContainerFactory listenerContainerFactory,
  28.             ConnectionFactory connectionFactory) {
  29.         this.listenerContainerFactory = listenerContainerFactory;
  30.         this.connectionFactory = connectionFactory;
  31.     }
  33.     /**
  34.      * 创建监听器容器并启动监听
  35.      *
  36.      * @param queueName 队列名称
  37.      */
  38.     public void createListener(String queueName) {
  39.         // 创建并启动监听器容器
  40.         SimpleMessageListenerContainer container = listenerContainerFactory.createListenerContainer();
  41.         container.setConnectionFactory(connectionFactory);
  42.         container.setQueueNames(queueName);
  44.         // 监听逻辑处理
  45.         container.setMessageListener(new MessageListenerAdapter(new Object() {
  46.             public void handleMessage(String message) {
  47.                 System.out.println("收到来自RabbitMQ中队列：" + queueName + " 队列的消息：" + message);
  48.             }
  49.         }));
  51.         // 启动监听器容器
  52.         container.start();
  53.         System.out.println("RabbitMQ队列监听器已启动：" + queueName);
  54.     }
  55. }
复制代码
4、RabbitMQ中的Exchange、Queue动态操作接口

RabbitDynamicChannelController.java

  1. import com.chain.air.rpp.exchange.rabbit.RabbitDynamicConfigService;
  2. import jakarta.annotation.Resource;
  3. import org.springframework.web.bind.annotation.GetMapping;
  4. import org.springframework.web.bind.annotation.RequestMapping;
  5. import org.springframework.web.bind.annotation.RequestParam;
  6. import org.springframework.web.bind.annotation.RestController;
  8. import java.util.HashMap;
  9. import java.util.Map;
  11. /**
  12. * @className: RabbitDynamicController
  13. * @program: chain
  14. * @description: RabbitMQ 动态创建队列、交换机，绑定等操作
  15. * @author: kenny
  16. * @create: 2024-10-04 00:22
  17. * @version: 1.0.0
  18. */
  19. @RestController
  20. @RequestMapping("/rabbit/dynamic/channel")
  21. public class RabbitDynamicChannelController {
  23.     /**
  24.      * 动态创建队列和交换机
  25.      */
  26.     @Resource
  27.     private RabbitDynamicConfigService rabbitDynamicConfigService;
  29.     /**
  30.      * 动态创建队列
  31.      *
  32.      * @param queueName 队列名称
  33.      * @return 处理结果
  34.      */
  35.     @GetMapping("/createQueue")
  36.     public String createQueue(@RequestParam("queueName") String queueName) {
  37.         rabbitDynamicConfigService.createQueue(queueName);
  38.         return "队列已创建: " + queueName;
  39.     }
  41.     /**
  42.      * 动态创建交换机
  43.      *
  44.      * @param exchangeName 交换机名称
  45.      * @return 处理结果
  46.      */
  47.     @GetMapping("/createExchange")
  48.     public String createExchange(@RequestParam("exchangeName") String exchangeName) {
  49.         rabbitDynamicConfigService.createExchange(exchangeName);
  50.         return "交换机已创建: " + exchangeName;
  51.     }
  53.     /**
  54.      * 动态绑定队列和交换机
  55.      *
  56.      * @param queueName    队列名称
  57.      * @param exchangeName 交换机名称
  58.      * @param routingKey   路由键
  59.      * @return 处理结果
  60.      */
  61.     @GetMapping("/bindQueue")
  62.     public String bindQueueToExchange(@RequestParam("queueName") String queueName,
  63.                                       @RequestParam("exchangeName") String exchangeName,
  64.                                       @RequestParam("routingKey") String routingKey) {
  65.         rabbitDynamicConfigService.bindQueueToExchange(queueName, exchangeName, routingKey);
  66.         return "队列和交换机已绑定: " + queueName + " -> " + exchangeName;
  67.     }
  69.     /**
  70.      * 动态删除队列
  71.      *
  72.      * @param queueName 队列名称
  73.      * @return 处理结果
  74.      */
  75.     @GetMapping("/deleteQueue")
  76.     public String deleteQueue(@RequestParam("queueName") String queueName) {
  77.         rabbitDynamicConfigService.deleteQueue(queueName);
  78.         return "队列已删除: " + queueName;
  79.     }
  81.     /**
  82.      * 动态删除交换机
  83.      *
  84.      * @param exchangeName 交换机名称
  85.      * @return 处理结果
  86.      */
  87.     @GetMapping("/deleteExchange")
  88.     public String deleteExchange(@RequestParam("exchangeName") String exchangeName) {
  89.         rabbitDynamicConfigService.deleteExchange(exchangeName);
  90.         return "交换机已删除: " + exchangeName;
  91.     }
  93.     // 创建并绑定 Fanout 交换机
  94.     @GetMapping("/createDirectExchange")
  95.     public String createDirectExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam String routingKey) {
  96.         rabbitDynamicConfigService.createDirectExchange(exchangeName);
  97.         rabbitDynamicConfigService.bindQueueToDirectExchange(queueName, exchangeName, routingKey);
  98.         return "Fanout Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with routing key: " + routingKey;
  99.     }
  101.     // 创建并绑定 Fanout 交换机
  102.     @GetMapping("/createFanoutExchange")
  103.     public String createFanoutExchange(@RequestParam String exchangeName, @RequestParam String queueName) {
  104.         rabbitDynamicConfigService.createFanoutExchange(exchangeName);
  105.         rabbitDynamicConfigService.bindQueueToFanoutExchange(queueName, exchangeName);
  106.         return "Fanout Exchange and Queue Binding created: " + exchangeName + " -> " + queueName;
  107.     }
  109.     // 创建并绑定 Topic 交换机
  110.     @GetMapping("/createTopicExchange")
  111.     public String createTopicExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam String routingKey) {
  112.         rabbitDynamicConfigService.createTopicExchange(exchangeName);
  113.         rabbitDynamicConfigService.bindQueueToTopicExchange(queueName, exchangeName, routingKey);
  114.         return "Topic Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with routing key: " + routingKey;
  115.     }
  117.     // 创建并绑定 Headers 交换机
  118.     @GetMapping("/createHeadersExchange")
  119.     public String createHeadersExchange(@RequestParam String exchangeName, @RequestParam String queueName, @RequestParam Map<String, String> headersMap) {
  120.         Map<String, Object> headers = new HashMap<>(headersMap);
  121.         rabbitDynamicConfigService.createHeadersExchange(exchangeName);
  122.         rabbitDynamicConfigService.bindQueueToHeadersExchange(queueName, exchangeName, headers);
  123.         return "Headers Exchange and Queue Binding created: " + exchangeName + " -> " + queueName + " with headers: " + headers;
  124.     }
  125. }
复制代码
5、RabbitMQ中的Queue消息监听动态操作接口

RabbitChannelListenerController.java

  1. import com.chain.air.rpp.exchange.rabbit.RabbitDynamicConfigService;
  2. import jakarta.annotation.Resource;
  3. import org.springframework.web.bind.annotation.GetMapping;
  4. import org.springframework.web.bind.annotation.RequestMapping;
  5. import org.springframework.web.bind.annotation.RequestParam;
  6. import org.springframework.web.bind.annotation.RestController;
  8. /**
  9. * @className: RabbitListenerController
  10. * @program: chain
  11. * @description: RabbitMQ 监听器 Controller 组件
  12. * @author: kenny
  13. * @create: 2024-10-04 01:30
  14. * @version: 1.0.0
  15. */
  16. @RestController
  17. @RequestMapping("/rabbit/channel/listener")
  18. public class RabbitChannelListenerController {
  20.     @Resource
  21.     private RabbitDynamicConfigService rabbitDynamicConfigService;
  23.     /**
  24.      * 创建监听器，监听指定队列
  25.      *
  26.      * @param queueName 队列名称
  27.      * @return 处理结果
  28.      */
  29.     @GetMapping("/queue")
  30.     public String listenQueue(@RequestParam("queueName") String queueName) {
  31.         rabbitDynamicConfigService.createQueue(queueName, true);
  32.         return "开始监听队列：" + queueName;
  33.     }
  34. }
复制代码
三、动态exchange、queue的测试

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

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

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

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

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

盛世宏图

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表