devbox
运维做开发
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

多个springboot整合使用rabbitmq(使用注解的方式)_springboot 配置多个rabbitmq

作者：运维做开发 | 2024-07-05 04:34:54

赞

踩

springboot 配置多个rabbitmq

一、简述

先参考单个springboot使用rabbitmq和了解rabbitmq的五种模式

单个springboot整合rabbitmq_java-zh的博客-CSDN博客

二、创建项目

1、先创建两个springboot项目，一个做生产者，一个做消费者

 2、导包(生产者和消费者对应的内容都是一样)

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.4.1</version>
  9.         <relativePath/>
  10.     </parent>
  11.     <groupId>com.mq</groupId>
  12.     <artifactId>mqcloud</artifactId>
  13.     <version>0.0.1-SNAPSHOT</version>
  14.     <properties>
  15.         <spring-cloud-alibaba.version>2021.1</spring-cloud-alibaba.version>
  16.         <java.version>1.8</java.version>
  17.         <spring-boot.version>2.5.3</spring-boot.version>
  18.         <lombok.version>1.18.14</lombok.version>
  19.         <mybaits.plus.version>3.3.2</mybaits.plus.version>
  20.         <alibaba.json.version>1.2.75</alibaba.json.version>
  21.         <spring-cloud-starter-bootstrap.version>3.0.2</spring-cloud-starter-bootstrap.version>
  22.     </properties>
  23.  
  24.     <!--    <dependencyManagement>-->
  25.     <dependencies>
  26.         <dependency>
  27.             <groupId>org.springframework.boot</groupId>
  28.             <artifactId>spring-boot-starter-web</artifactId>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-starter-actuator</artifactId>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.springframework.boot</groupId>
  36.             <artifactId>spring-boot-devtools</artifactId>
  37.         </dependency>
  38.         <dependency>
  39.             <groupId>org.springframework.boot</groupId>
  40.             <artifactId>spring-boot-starter-test</artifactId>
  41.             <scope>test</scope>
  42.         </dependency>
  43.         <!--AOP-->
  44.         <dependency>
  45.             <groupId>org.springframework.boot</groupId>
  46.             <artifactId>spring-boot-starter-aop</artifactId>
  47.         </dependency>
  48.         <!--bootstrap 启动器-->
  49.         <dependency>
  50.             <groupId>org.springframework.cloud</groupId>
  51.             <artifactId>spring-cloud-starter-bootstrap</artifactId>
  52.             <version>${spring-cloud-starter-bootstrap.version}</version>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>com.alibaba</groupId>
  56.             <artifactId>fastjson</artifactId>
  57.             <version>${alibaba.json.version}</version>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>org.projectlombok</groupId>
  61.             <artifactId>lombok</artifactId>
  62.             <version>${lombok.version}</version>
  63.             <optional>true</optional>
  64.         </dependency>
  65.         <!--jjwt-->
  66.         <dependency>
  67.             <groupId>io.jsonwebtoken</groupId>
  68.             <artifactId>jjwt</artifactId>
  69.             <version>0.9.0</version>
  70.         </dependency>
  71.         <dependency>
  72.             <groupId>com.google.guava</groupId>
  73.             <artifactId>guava</artifactId>
  74.             <version>31.1-jre</version>
  75.         </dependency>
  76.         <dependency>
  77.             <groupId>com.alibaba.cloud</groupId>
  78.             <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
  79.             <version>${spring-cloud-alibaba.version}</version>
  80.         </dependency>
  81.         <dependency>
  82.             <groupId>com.alibaba.cloud</groupId>
  83.             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
  84.             <version>${spring-cloud-alibaba.version}</version>
  85.         </dependency>
  86.  
  87.  
  88.         <dependency>
  89.             <groupId>org.apache.commons</groupId>
  90.             <artifactId>commons-lang3</artifactId>
  91.             <version>3.12.0</version>
  92.         </dependency>
  93.         <!--zhong 基础jar-->
  94.         <dependency>
  95.             <groupId>com.common</groupId>
  96.             <artifactId>base</artifactId>
  97.             <version>1.0-SNAPSHOT</version>
  98.         </dependency>
  99.  
  100.         <!--消息队列 -->
  101.         <dependency>
  102.             <groupId>org.springframework.boot</groupId>
  103.             <artifactId>spring-boot-starter-amqp</artifactId>
  104.             <version>3.0.4</version>
  105.         </dependency>
  106.     </dependencies>
  107.     <!--    </dependencyManagement>-->
  108.  
  109.  
  110.  
  111. </project>

3、编写配置文件

这里的配置文件生产者和消费者都一样

  1. spring:
  2.  rabbitmq:
  3.     #如果是集群，用,隔开
  4.     connection-timeout: 15000
  5.     username: admin
  6.     password: 123456
  7.     #虚拟host，可以不设置
  8.     virtual-host: /
  9.     listener:
  10.       simple:
  11.         acknowledge-mode: auto  #manual:手动处理 auto:自动处理
  12.         #消费端监听个数(即@RabbitListenter开启几个线程去处理)
  13.         concurrency: 10
  14.         #消费端监听的最大个数
  15.         max-concurrency: 10
  16.         prefetch: 5
  17.         default-requeue-rejected: true  #消费不成功的消息,拒绝入队
  18.         retry:
  19.           enabled: true #开启消息重试
  20.           max-attempts: 4 #重试次数
  21.           max-interval: 10000 #重试最大间隔时间
  22.           initial-interval: 2000 #重试初始间隔时间
  23.     #消息确认方式,通过correlated来确认
  24.     publisher-confirm-type: correlated
  25.     publisher-returns: true
  26.     host: 192.168.139.128
  27.     port: 5672
  28. rabbitmq:
  29.   #订阅模式(扇形模式)  
  30.   fanout:
  31.     exchange:
  32.       #交换机名称
  33.       name1: exchange.fanout
  34.     queue:
  35.       #通道名称
  36.       name1: exchange.fanout.queue_1
  37.       name2: exchange.fanout.queue_2
  38.   #交换机模式  
  39.   direct:
  40.     exchange:
  41.       name1: exchange.direct
  42.     queue:
  43.       name1: exchange.direct.queue_1
  44.       name2: exchange.direct.queue_2
  45.     routing:
  46.       name1: exchange.direct.routing.1
  47.       name2: exchange.direct.routing.2
  48.   #主题模式  
  49.   topic:
  50.     exchange: 
  51.       name1: exchange.topic
  52.     queue:
  53.       name1: exchange.topic.queue_1
  54.       name2: exchange.topic.queue_2
  55.     routing:
  56.       name1: '#.routingkey.#'
  57.       name2: routingkey.*

三、编码代码和测试结果

这里以rabbitmq的订阅模式(扇形模式)、路由器模式、主题模式为案例

3.1 扇形模式

生产者

  1. @Autowired
  2. private  RabbitTemplate rabbitTemplate;
  3.  
  4. // 订阅模式的交换机名称
  5. @Value("${rabbitmq.fanout.exchange.name1}")
  6.  private String exchangeName;
  7.  
  8.  
  9. /**
  10.   * 订阅模式(扇形模式)生产者
  11.   *
  12.   * @param context
  13.   */
  14. @GetMapping("/fanout/{context}")
  15. public void sendMessage(@PathVariable String context) {
  16.    System.out.println("需要发送得内容为:" + context);
  17.    rabbitTemplate.convertAndSend(exchangeName, "", context);
  18. }

消费者

注意：消费者绑定交换机和通道的值必须是固定常量值，所以我们直接从配置文件中读取

  1. @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = "${rabbitmq.fanout.exchange.name1}", type = ExchangeTypes.FANOUT),
  2.             value = @Queue(value = "${rabbitmq.fanout.queue.name1}")))
  3.     @RabbitHandler
  4.     public void consumerFanoutExchange(String context) {
  5.         System.out.println("订阅模式，通道一接收到的内容为内容:" + context);
  6.     }
  7.  
  8. @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = "${rabbitmq.fanout.exchange.name1}", type = ExchangeTypes.FANOUT),
  9.             value = @Queue(value = "${rabbitmq.fanout.queue.name2}")))
  10.     @RabbitHandler
  11.     public void consumerFanoutExchange2(String context) {
  12.         System.out.println("订阅模式，通道二接收到的内容为内容:" + context);
  13. }

3.2 路由器模式

生产者

  1. @Autowired
  2. private  RabbitTemplate rabbitTemplate;
  3.  
  4. // 路由器模式的交换机名称
  5. @Value("${rabbitmq.direct.exchange.name1}")
  6. private String directName;
  7.  
  8. // 路由器模式的路由key1
  9. @Value("${rabbitmq.direct.routing.name1}")
  10. private String directRoutingKeyName1;
  11.  
  12. //路由器模式的路由key2
  13. @Value("${rabbitmq.direct.routing.name2}")
  14. private String directRoutingKeyName2;
  15.  
  16. /**
  17.   * 路由器模式
  18.   * @param context
  19.   * @param routingkey
  20.   */
  21. @GetMapping("/direct")
  22. public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
  23.    if (1 == routingkey) {
  24.       rabbitTemplate.convertAndSend(directName, directRoutingKeyName1, context);
  25.    } else if (2 == routingkey) {
  26.       rabbitTemplate.convertAndSend(directName, directRoutingKeyName2, context);
  27.    } else {
  28.       System.out.println("数据非法");
  29.    }
  30. }

消费者

  1.  
  2.     @RabbitListener(bindings = @QueueBinding(
  3.             exchange = @Exchange(value = "${rabbitmq.direct.exchange.name1}", type = ExchangeTypes.DIRECT),
  4.             value = @Queue(value = "${rabbitmq.direct.queue.name1}"),
  5.             key = "${rabbitmq.direct.routing.name1}"))
  6.     public void exchangeDirectRoutingKey1(String context, Message message) {
  7.         System.out.println("key1:" + message.getMessageProperties().getReceivedRoutingKey());
  8.         System.out.println("路由器模式1 接收到的消息为:" + context);
  9.     }
  10.  
  11.     @RabbitListener(bindings = @QueueBinding(
  12.             exchange = @Exchange(value = "${rabbitmq.direct.exchange.name1}", type = ExchangeTypes.DIRECT),
  13.             value = @Queue(value = "${rabbitmq.direct.queue.name2}"),
  14.             key = "${rabbitmq.direct.routing.name2}"))
  15.     public void exchangeDirectRoutingKey2(String context, Message message) {
  16.         System.out.println("key2:" + message.getMessageProperties().getReceivedRoutingKey());
  17.         System.out.println("路由器模式2 接收到的消息为" + context);
  18.     }

3.3 主题模式

生产者

  1. // 主题模式的交换机名称
  2.     @Value("${rabbitmq.topic.exchange.name1}")
  3.     private String topicName;
  4.  
  5.     //用来匹配主题模式对应的key
  6.     public static final String EXCHANGE_TOPIC_CASE_KEY_1 = "topic.routingkey.case1";
  7.  
  8.     //如果case_key_2这样写,那么绑定case_key_1的队列一样会接收到，因为case_key_2也一样和key1匹配上
  9.     public static final String EXCHANGE_TOPIC_CASE_KEY_2 = "routingkey.case2";
  10.  
  11.     @GetMapping("/topic")
  12.     public void sendMessageTopic(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
  13.         if (1 == routingkey) {
  14.             rabbitTemplate.convertAndSend(topicName, EXCHANGE_TOPIC_CASE_KEY_1, context + routingkey);
  15.         } else if (2 == routingkey) {
  16.             rabbitTemplate.convertAndSend(topicName, EXCHANGE_TOPIC_CASE_KEY_2, context + routingkey);
  17.         } else {
  18.             System.out.println("数据非法");
  19.         }
  20.     }

消费者

  1. @RabbitListener(bindings = @QueueBinding(
  2.             exchange = @Exchange(value = "${rabbitmq.topic.exchange.name1",  type = ExchangeTypes.TOPIC),
  3.             value = @Queue(value = "${rabbitmq.topic.queue.name1}"),
  4.             key = "${rabbitmq.topic.routing.name1}"))
  5.     @RabbitHandler
  6.     public void exchangeTopicRoutingKey1(String context, Message message) {
  7.         System.out.println("key1:"+message.getMessageProperties().getReceivedRoutingKey());
  8.         System.out.println("主题模式1:接收的内容为:"+ context);
  9.     }
  10.  
  11. @RabbitListener(bindings = @QueueBinding(
  12.             exchange = @Exchange(value = "${rabbitmq.topic.exchange.name1", type = ExchangeTypes.TOPIC),
  13.             value = @Queue(value = "${rabbitmq.topic.queue.name2}"),
  14.             key = "${rabbitmq.topic.routing.name2}"))
  15.     @RabbitHandler
  16.     public void exchangeTopicRoutingKey2(String context,  Message message) {
  17.         System.out.println("key2:"+message.getMessageProperties().getReceivedRoutingKey());
  18.         System.out.println("主题模式2:接收的内容为:"+ context);
  19.     }

3.4 全量代码

生产者

  1. @RestController
  2. @RequiredArgsConstructor
  3. @RequestMapping("/test/mq")
  4. public class ProductController {
  5.  
  6.     private final RabbitTemplate rabbitTemplate;
  7.  
  8.     // 订阅模式的交换机名称
  9.     @Value("${rabbitmq.fanout.exchange.name1}")
  10.     private String exchangeName;
  11.  
  12.     // 路由器模式的交换机名称
  13.     @Value("${rabbitmq.direct.exchange.name1}")
  14.     private String directName;
  15.  
  16.     // 路由器模式的路由key1
  17.     @Value("${rabbitmq.direct.routing.name1}")
  18.     private String directRoutingKeyName1;
  19.  
  20.     //路由器模式的路由key2
  21.     @Value("${rabbitmq.direct.routing.name2}")
  22.     private String directRoutingKeyName2;
  23.  
  24.     // 主题模式的交换机名称
  25.     @Value("${rabbitmq.topic.exchange.name1}")
  26.     private String topicName;
  27.  
  28.     //用来匹配主题模式对应的key
  29.     public static final String EXCHANGE_TOPIC_CASE_KEY_1 = "topic.routingkey.case1";
  30.  
  31.     //如果case_key_2这样写,那么绑定case_key_1的队列一样会接收到，因为case_key_2也一样和key1匹配上
  32.     public static final String EXCHANGE_TOPIC_CASE_KEY_2 = "routingkey.case2";
  33.  
  34.  
  35.     /**
  36.      * 订阅模式(扇形模式)生产者
  37.      *
  38.      * @param context
  39.      */
  40.     @GetMapping("/fanout/{context}")
  41.     public void sendMessage(@PathVariable String context) {
  42.         System.out.println("需要发送得内容为:" + context);
  43.         rabbitTemplate.convertAndSend(exchangeName, "", context);
  44.     }
  45.  
  46.     /**
  47.      * 路由器模式
  48.      * @param context
  49.      * @param routingkey
  50.      */
  51.     @GetMapping("/direct")
  52.     public void sendMessageDirect(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
  53.         if (1 == routingkey) {
  54.             rabbitTemplate.convertAndSend(directName, directRoutingKeyName1, context);
  55.         } else if (2 == routingkey) {
  56.             rabbitTemplate.convertAndSend(directName, directRoutingKeyName2, context);
  57.         } else {
  58.             System.out.println("数据非法");
  59.         }
  60.     }
  61.  
  62.     @GetMapping("/topic")
  63.     public void sendMessageTopic(@RequestParam("context") String context, @RequestParam("routingkey") Integer routingkey) {
  64.         if (1 == routingkey) {
  65.             rabbitTemplate.convertAndSend(topicName, EXCHANGE_TOPIC_CASE_KEY_1, context + routingkey);
  66.         } else if (2 == routingkey) {
  67.             rabbitTemplate.convertAndSend(topicName, EXCHANGE_TOPIC_CASE_KEY_2, context + routingkey);
  68.         } else {
  69.             System.out.println("数据非法");
  70.         }
  71.     }
  72. }

消费者

  1. @Component
  2. public class ConsumerTest {
  3.  
  4.  
  5.     @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = "${rabbitmq.fanout.exchange.name1}", type = ExchangeTypes.FANOUT),
  6.             value = @Queue(value = "${rabbitmq.fanout.queue.name1}")))
  7.     @RabbitHandler
  8.     public void consumerFanoutExchange(String context) {
  9.         System.out.println("订阅模式，通道一接收到的内容为内容:" + context);
  10.     }
  11.  
  12.     @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = "${rabbitmq.fanout.exchange.name1}", type = ExchangeTypes.FANOUT),
  13.             value = @Queue(value = "${rabbitmq.fanout.queue.name2}")))
  14.     @RabbitHandler
  15.     public void consumerFanoutExchange2(String context) {
  16.         System.out.println("订阅模式，通道二接收到的内容为内容:" + context);
  17.     }
  18.  
  19.  
  20.     @RabbitListener(bindings = @QueueBinding(
  21.             exchange = @Exchange(value = "${rabbitmq.direct.exchange.name1}", type = ExchangeTypes.DIRECT),
  22.             value = @Queue(value = "${rabbitmq.direct.queue.name1}"),
  23.             key = "${rabbitmq.direct.routing.name1}"))
  24.     public void exchangeDirectRoutingKey1(String context, Message message) {
  25.         System.out.println("key1:" + message.getMessageProperties().getReceivedRoutingKey());
  26.         System.out.println("路由器模式1 接收到的消息为:" + context);
  27.     }
  28.  
  29.     @RabbitListener(bindings = @QueueBinding(
  30.             exchange = @Exchange(value = "${rabbitmq.direct.exchange.name1}", type = ExchangeTypes.DIRECT),
  31.             value = @Queue(value = "${rabbitmq.direct.queue.name2}"),
  32.             key = "${rabbitmq.direct.routing.name2}"))
  33.     public void exchangeDirectRoutingKey2(String context, Message message) {
  34.         System.out.println("key2:" + message.getMessageProperties().getReceivedRoutingKey());
  35.         System.out.println("路由器模式2 接收到的消息为" + context);
  36.     }
  37.  
  38.     @RabbitListener(bindings = @QueueBinding(
  39.             exchange = @Exchange(value = "${rabbitmq.topic.exchange.name1",  type = ExchangeTypes.TOPIC),
  40.             value = @Queue(value = "${rabbitmq.topic.queue.name1}"),
  41.             key = "${rabbitmq.topic.routing.name1}"))
  42.     @RabbitHandler
  43.     public void exchangeTopicRoutingKey1(String context, Message message) {
  44.         System.out.println("key1:"+message.getMessageProperties().getReceivedRoutingKey());
  45.         System.out.println("主题模式1:接收的内容为:"+ context);
  46.     }
  47.  
  48.     @RabbitListener(bindings = @QueueBinding(
  49.             exchange = @Exchange(value = "${rabbitmq.topic.exchange.name1", type = ExchangeTypes.TOPIC),
  50.             value = @Queue(value = "${rabbitmq.topic.queue.name2}"),
  51.             key = "${rabbitmq.topic.routing.name2}"))
  52.     @RabbitHandler
  53.     public void exchangeTopicRoutingKey2(String context,  Message message) {
  54.         System.out.println("key2:"+message.getMessageProperties().getReceivedRoutingKey());
  55.         System.out.println("主题模式2:接收的内容为:"+ context);
  56.     }
  57.  
  58.  
  59. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/运维做开发/article/detail/789213
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号