2024/5/13 SpringBoot配置多个RabbitMQ

目录
 一、单个RabbitMQ配置
        1.1、导入Maven坐标
        1.2、yaml配置
        1.3、java配置类
                1.3.1、交换机配置
                1.3.2、队列配置
                1.3.3、绑定配置
                1.3.4、连接配置
        1.4、生产者与消费者利用配置
                1.4.1、生产者利用配置
                1.4.2、消费者利用配置
二、多个RabbitMQ配置
        2.1、yaml配置 
        2.2、java配置类
         2.3、生产者与消费者利用配置
                2.3.1、生产者利用配置
                2.3.1、消费者利用配置
三、总结 

---

   需求形貌：原SpringBoot工程已经配置了一个RabbitMQ，现需求是再配置一个RabbitMQ，实现效果是不同RabbitMQ推送到不同的队列中，且互不干扰影响使用。
   一、单个RabbitMQ配置

        1.1、导入Maven坐标

1.         <dependency>
2.             <groupId>com.rabbitmq</groupId>
3.             <artifactId>amqp-client</artifactId>
4.             <version>5.10.0</version>
5.         </dependency>
6.         <dependency>
7.             <groupId>org.springframework.boot</groupId>
8.             <artifactId>spring-boot-starter-amqp</artifactId>
9.             <version>2.4.4</version>
10.         </dependency>

复制代码

        1.2、yaml配置

1.   rabbitmq:
2.     host: xx.xxx.xxx.xxx
3.     port: xxxx
4.     username: xxxx
5.     password: xxxxxx
6.     virtual-host: xxxx
7.     publisher-returns: true
8.     publisher-confirms: true
9.     listener:
10.        simple:
11.          default-requeue-rejected: true
12.            retry:
13.             enabled: false
14.             max-attempts: 3
15.             initial-interval: 5000

复制代码

        1.3、java配置类

                1.3.1、交换机配置

1. package com.ruoyi.report.config;
3. import org.springframework.amqp.core.DirectExchange;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.stereotype.Component;
7. @Component
8. public class ExchangeConfig {
10.     public static final String ecoa_exchange = "ecoaExchange";
12.     /**
13.      * 1.定义direct exchange
14.      * 2.durable="true" rabbitmq重启的时候不需要创建新的交换机
15.      * 3.direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列
16.      */
17.     @Bean
18.     public DirectExchange ecoaExchange() {
19.         DirectExchange directExchange = new DirectExchange(ecoa_exchange, true, false);
20.         return directExchange;
21.     }
23. }

复制代码

                1.3.2、队列配置

1. package com.ruoyi.report.config;
3. import org.springframework.amqp.core.Queue;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.stereotype.Component;
7. /**
8. * @ClassName QueueConfig
9. * @Description
10. * @Author Mr.Huang
11. * @Date 2023/9/22 16:26
12. * @Version 1.0
13. **/
14. @Component
15. public class QueueConfig {
17.     private static final String ecoa_file_upload_queue = "ecoa_file_upload_queue";
19.     @Bean
20.     public Queue ecoaFileUploadDispatchQueue() {
21.         /**
22.          durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
23.          auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
24.          exclusive  表示该消息队列是否只在当前connection生效,默认是false
25.          */
26.         return new Queue(ecoa_file_upload_queue, true, false, false);
27.     }
28. }

复制代码

                1.3.3、绑定配置

1. package com.ruoyi.report.config;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.beans.factory.annotation.Autowired;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.stereotype.Component;
9. /**
10. * @ClassName BindingConfig
11. * @Description
12. * @Author Mr.Huang
13. * @Date 2023/9/22 16:31
14. * @Version 1.0
15. **/
16. @Component
17. public class BindingConfig {
18.     @Autowired
19.     private QueueConfig queueConfig;
20.     @Autowired
21.     private ExchangeConfig exchangeConfig;
23.     public static final String ECOA_file_upload_key = "ecoa_file_upload_key";
26.     @Bean
27.     public Binding ecoaFileUploadDispatchBinding() {
28.         return BindingBuilder.bind(queueConfig.ecoaFileUploadDispatchQueue()).to(exchangeConfig.ecoaExchange()).with(ECOA_file_upload_key);
29.     }
30. }

复制代码

                1.3.4、连接配置

1. package com.ruoyi.report.config;
3. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
5. import org.springframework.beans.factory.annotation.Autowired;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. /**
10. * @ClassName RabbitMqConfig
11. * @Description
12. * @Author Mr.Huang
13. * @Date 2023/9/22 16:14
14. * @Version 1.0
15. **/
16. @Configuration
17. public class RabbitMqConfig {
18.     /**
19.      * 连接工厂
20.      */
21.     @Autowired
22.     private ConnectionFactory connectionFactory;
24.     /**
25.      * 自定义rabbit template用于数据的接收和发送
26.      * 可以设置消息确认机制和回调
27.      *
28.      * @return
29.      */
30.     @Bean
31.     public RabbitTemplate rabbitTemplate() {
32.         RabbitTemplate template = new RabbitTemplate(connectionFactory);
33.         return template;
34.     }
35. }

复制代码

        1.4、生产者与消费者利用配置

                1.4.1、生产者利用配置

1. package com.ruoyi.report.utils;
3. import com.alibaba.fastjson.JSON;
4. import com.alibaba.fastjson.serializer.SerializerFeature;
5. import com.ruoyi.common.utils.StringUtils;
6. import com.ruoyi.report.config.BindingConfig;
7. import com.ruoyi.report.config.ExchangeConfig;
8. import org.springframework.amqp.core.Message;
9. import org.springframework.amqp.core.MessageBuilder;
10. import org.springframework.amqp.core.MessageProperties;
11. import org.springframework.amqp.rabbit.connection.CorrelationData;
12. import org.springframework.amqp.rabbit.core.RabbitTemplate;
13. import org.springframework.beans.factory.annotation.Autowired;
14. import org.springframework.stereotype.Component;
16. import java.util.UUID;
18. /**
19. * @ClassName MessageUtils
20. * @Description
21. * @Author Mr.Huang
22. * @Date 2023/9/22 16:36
23. * @Version 1.0
24. **/
25. @Component
26. public class MessageUtils {
27.     @Autowired
28.     private RabbitTemplate rabbitTemplate;
30.     /**
31.      * 发送消息
32.      * 发送随货单信息
33.      * @param message 消息
34.      */
35.     public void sendMessage(Object message) {
36.         String uuid = UUID.randomUUID().toString();
37.         CorrelationData correlationId = new CorrelationData(uuid);
38.         Message msg = MessageBuilder.withBody(JSON.toJSONStringWithDateFormat(message, "yyyy-MM-dd HH:mm:ss", SerializerFeature.WriteNullStringAsEmpty).getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();
39.         rabbitTemplate.convertAndSend(ExchangeConfig.ecoa_exchange, BindingConfig.ECOA_file_upload_key, msg, correlationId);
40.     }
41. }

复制代码

                1.4.2、消费者利用配置

1. package com.ruoyi.report.consumer;
3. import com.alibaba.fastjson.JSONObject;
4. import com.rabbitmq.client.Channel;
5. import com.ruoyi.report.config.RabbitMqConfig;
6. import com.ruoyi.report.entity.open.PrintResult;
7. import com.ruoyi.report.service.open.PrintSendLogService;
8. import lombok.extern.slf4j.Slf4j;
9. import org.springframework.amqp.core.Message;
10. import org.springframework.amqp.rabbit.annotation.RabbitListener;
11. import org.springframework.beans.factory.annotation.Autowired;
12. import org.springframework.stereotype.Component;
14. import java.io.IOException;
16. /**
17. * @ClassName PrintFeedbackConsumer
18. * @Description
19. * @Author Mr.Huang
20. * @Date 2024/4/30 10:23
21. * @Version 1.0
22. **/
23. @Slf4j
24. @Component
25. public class PrintFeedbackConsumer {
27.     @Autowired
28.     private PrintSendLogService printSendLogService;
30.     @RabbitListener(queues = {RabbitMqConfig.print_4pl_dispatch_info_feedback_queue}, containerFactory = "printContainerFactory")
31.     public void receiveMq(Message message, Channel channel) {
32.         try {
33.             String body = new String(message.getBody());
34.             log.info("接受【Print结果推送】RabbitMQ消息："+body);
35.             JSONObject objJson = JSONObject.parseObject(body);
36.             Thread.sleep(1000);
37.             PrintResult printResult = JSONObject.toJavaObject(objJson, PrintResult.class);
38.             printSendLogService.updatePrintSendLog(printResult);
39.         }catch (Exception e){
40.             log.error("",e);
41.             try {
42.                 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
43.             } catch (IOException ex) {
44.                 ex.printStackTrace();
45.             }
46.         }
47.     }
48. }

复制代码

二、多个RabbitMQ配置

           Maven坐标与上面单个RabbitMQ配置同等
          2.1、yaml配置 

1.   rabbitmq:
2.     first:
3.       host: xx.xxx.xxx.xxx
4.       port: xxxx
5.       username: xxxx
6.       password: xxxxxx
7.       virtual-host: xxxx
8.       publisher-returns: true
9.       publisher-confirms: true
10.       listener:
11.           simple:
12.             default-requeue-rejected: true
13.             retry:
14.               enabled: false
15.               max-attempts: 3
16.               initial-interval: 5000         
17.     second:
18.       host: xx.xxx.xxx.xxx
19.       port: xxxx
20.       username: xxxx
21.       password: xxxxxx
22.       publisher-returns: true
23.       publisher-confirms: true
24.       virtual-host: xxxx
25.       listener:
26.           simple:
27.             default-requeue-rejected: true
28.             retry:
29.               enabled: false
30.               max-attempts: 3
31.               initial-interval: 5000

复制代码

        2.2、java配置类

1. package com.ruoyi.report.config;
3. import com.rabbitmq.client.Channel;
4. import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
5. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
6. import org.springframework.amqp.rabbit.connection.Connection;
7. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
8. import org.springframework.amqp.rabbit.core.RabbitTemplate;
9. import org.springframework.beans.factory.annotation.Autowired;
10. import org.springframework.beans.factory.annotation.Qualifier;
11. import org.springframework.beans.factory.annotation.Value;
12. import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
13. import org.springframework.context.annotation.Bean;
14. import org.springframework.context.annotation.Configuration;
15. import org.springframework.context.annotation.Primary;
17. /**
18. * @ClassName RabbitMqConfig
19. * @Description
20. * @Author Mr.Huang
21. * @Date 2023/9/22 16:14
22. * @Version 1.0
23. **/
24. @Configuration
25. public class RabbitMqConfig {
26.     // 第一个MQ电子药检队列与key
27.     public static final String ECOA_file_upload_queue = "ecoa_file_upload_queue";
28.     public static final String ECOA_file_upload_key = "ecoa_file_upload_key";
30.     // 第二个MQ单据打印平台队列与key
31.     public static final String print_tms_dispatch_info_queue = "print_tms_dispatch_info_queue";
32.     public static final String print_4pl_dispatch_info_feedback_queue = "print_4pl_dispatch_info_feedback_queue";
33.     public static final String print_tms_dispatch_info_key = "print_tms_dispatch_info_key";
34.     public static final String print_4pl_dispatch_info_feedback_key = "print_4pl_dispatch_info_feedback_key";
35.     /** 交换机名称 */
36.     public static final String EXCHANGE = "ecoaExchange";
37.     public static final String EXCHANGE2 = "tms_exchange";
39.     /** 第一个rabbitMq队列 */
40.     @Bean(name = "ECOAConnectionFactory")
41.     @Primary
42.     public ConnectionFactory ECOAConnectionFactory(@Value("${spring.rabbitmq.first.host}") String host,
43.                                                   @Value("${spring.rabbitmq.first.port}") int port,
44.                                                   @Value("${spring.rabbitmq.first.username}") String username,
45.                                                   @Value("${spring.rabbitmq.first.password}") String password,
46.                                                   @Value("${spring.rabbitmq.first.virtual-host}") String virtualHost) {
47.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
48.         connectionFactory.setHost(host);
49.         connectionFactory.setPort(port);
50.         connectionFactory.setUsername(username);
51.         connectionFactory.setPassword(password);
52.         connectionFactory.setVirtualHost(virtualHost);
53.         return connectionFactory;
54.     }
56.     /** 第二个rabbitMq队列 */
57.     @Bean(name = "printConnectionFactory")
58.     public ConnectionFactory printConnectionFactory(@Value("${spring.rabbitmq.second.host}") String host,
59.                                                    @Value("${spring.rabbitmq.second.port}") int port,
60.                                                    @Value("${spring.rabbitmq.second.username}") String username,
61.                                                    @Value("${spring.rabbitmq.second.password}") String password,
62.                                                    @Value("${spring.rabbitmq.second.virtual-host}") String virtualHost) {
63.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
64.         connectionFactory.setHost(host);
65.         connectionFactory.setPort(port);
66.         connectionFactory.setUsername(username);
67.         connectionFactory.setPassword(password);
68.         connectionFactory.setVirtualHost(virtualHost);
69.         return connectionFactory;
70.     }
71.     /** 第一个rabbitMq操作模板 */
72.     @Bean(name="ECOARabbitTemplate")
73.     @Primary
74.     public RabbitTemplate fplRabbitTemplate(@Qualifier("ECOAConnectionFactory") ConnectionFactory connectionFactory){
75.         RabbitTemplate firstRabbitTemplate = new RabbitTemplate(connectionFactory);
76.         return firstRabbitTemplate;
77.     }
78.     /** 第二个rabbitMq操作模板 */
79.     @Bean(name="printRabbitTemplate")
80.     public RabbitTemplate tcscRabbitTemplate(@Qualifier("printConnectionFactory") ConnectionFactory connectionFactory){
81.         RabbitTemplate secondRabbitTemplate = new RabbitTemplate(connectionFactory);
82.         return secondRabbitTemplate;
83.     }
84.     /** 第一个rabbitMq连接工厂 */
85.     @Bean(name="ECOAContainerFactory")
86.     public SimpleRabbitListenerContainerFactory ECOAContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
87.                                                                     @Qualifier("ECOAConnectionFactory") ConnectionFactory connectionFactory) {
88.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
89.         factory.setMaxConcurrentConsumers(5);
90.         factory.setConcurrentConsumers(1);
91.         factory.setPrefetchCount(1);
92.         configurer.configure(factory, connectionFactory);
93.         return factory;
94.     }
95.     /** 第二个rabbitMq连接工厂 */
96.     @Bean(name="printContainerFactory")
97.     public SimpleRabbitListenerContainerFactory printContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
98.                                                                      @Qualifier("printConnectionFactory") ConnectionFactory connectionFactory) {
99.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
100.         factory.setMaxConcurrentConsumers(5);
101.         factory.setConcurrentConsumers(1);
102.         factory.setPrefetchCount(1);
103.         configurer.configure(factory, connectionFactory);
104.         return factory;
105.     }
106.     /** 第一个mq绑定队列绑定交换机 */
107.     @Bean
108.     public String runECOAQueue(@Qualifier("ECOAConnectionFactory") ConnectionFactory connectionFactory) {
109.         System.out.println("configuration ECOAQueue ........................");
110.         Connection connection = connectionFactory.createConnection();
111.         Channel channel = connection.createChannel(false);
112.         try {
113.             channel.exchangeDeclare(EXCHANGE, "direct", true, false, null);
114.             // 单据推送电子药检队列
115.             channel.queueDeclare(ECOA_file_upload_queue, true, false, false, null);
116.             channel.queueBind(ECOA_file_upload_queue, EXCHANGE, ECOA_file_upload_key);
118.         } catch (Exception e) {
119.             e.printStackTrace();
120.         } finally {
121.             return "ECOAQueue";
122.         }
123.     }
124.     /** 第二个mq绑定队列绑定交换机 */
125.     @Bean
126.     public String runPrintQueue(@Qualifier("printConnectionFactory") ConnectionFactory connectionFactory) {
127.         System.out.println("configuration printQueue ........................");
128.         Connection connection = connectionFactory.createConnection();
129.         Channel channel = connection.createChannel(false);
130.         try {
131.             channel.exchangeDeclare(EXCHANGE2, "direct", true, false, null);
132.             // 单据推送单据打印平台队列
133.             channel.queueDeclare(print_tms_dispatch_info_queue, true, false, false, null);
134.             channel.queueBind(print_tms_dispatch_info_queue, EXCHANGE2, print_tms_dispatch_info_key);
135.             // 单据打印平台反馈队列
136.             channel.queueDeclare(print_4pl_dispatch_info_feedback_queue,true,false,false,null);
137.             channel.queueBind(print_4pl_dispatch_info_feedback_queue,EXCHANGE2,print_4pl_dispatch_info_feedback_key);
139.         } catch (Exception e) {
140.             e.printStackTrace();
141.         } finally {
142.             return "printQueue";
143.         }
144.     }
145. }

复制代码

  留意：需将原MQ：交换机、队列、绑定配置类解释掉，只留这一个配置文件即可，这个配置文件已经将对应的：交换机、队列绑定好，只是需要留意队列名字、交换机不要绑定错了
           2.3、生产者与消费者利用配置

                2.3.1、生产者利用配置

1. package com.ruoyi.report.utils;
3. import com.alibaba.fastjson.JSON;
4. import com.alibaba.fastjson.serializer.SerializerFeature;
5. import com.ruoyi.report.config.RabbitMqConfig;
6. import org.springframework.amqp.core.Message;
7. import org.springframework.amqp.core.MessageBuilder;
8. import org.springframework.amqp.core.MessageProperties;
9. import org.springframework.amqp.rabbit.connection.CorrelationData;
10. import org.springframework.amqp.rabbit.core.RabbitTemplate;
11. import org.springframework.stereotype.Component;
13. import javax.annotation.Resource;
14. import java.util.UUID;
16. /**
17. * @ClassName MessageUtils
18. * @Description
19. * @Author Mr.Huang
20. * @Date 2023/9/22 16:36
21. * @Version 1.0
22. **/
23. @Component
24. public class MessageUtils {
26.     @Resource(name = "ECOARabbitTemplate")
27.     private RabbitTemplate ECOARabbitTemplate;
29.     @Resource(name = "printRabbitTemplate")
30.     private RabbitTemplate printRabbitTemplate;
32.     /**
33.      * 向ECOA发送消息
34.      * 发送随货单信息
35.      * @param message 消息
36.      */
37.     public void sendMessage(Object message) {
38.         String uuid = UUID.randomUUID().toString();
39.         CorrelationData correlationId = new CorrelationData(uuid);
40.         Message msg = MessageBuilder.withBody(JSON.toJSONStringWithDateFormat(message, "yyyy-MM-dd HH:mm:ss", SerializerFeature.WriteNullStringAsEmpty).getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();
41.         ECOARabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE, RabbitMqConfig.ECOA_file_upload_key, msg, correlationId);
42.     }
44.     /**
45.      * 向print发送消息
46.      * 发送派车单信息
47.      * @param message 消息
48.      */
49.     public void sendPrintMessage(Object message) {
50.         String uuid = UUID.randomUUID().toString();
51.         CorrelationData correlationId = new CorrelationData(uuid);
52.         Message msg = MessageBuilder.withBody(JSON.toJSONStringWithDateFormat(message, "yyyy-MM-dd HH:mm:ss", SerializerFeature.WriteNullStringAsEmpty).getBytes()).setContentType(MessageProperties.CONTENT_TYPE_JSON).build();
53.         printRabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE2, RabbitMqConfig.print_tms_dispatch_info_key, msg, correlationId);
54.     }
55. }

复制代码

                2.3.1、消费者利用配置

1. package com.ruoyi.report.consumer;
3. import com.alibaba.fastjson.JSONObject;
4. import com.rabbitmq.client.Channel;
5. import com.ruoyi.report.config.RabbitMqConfig;
6. import com.ruoyi.report.entity.open.PrintResult;
7. import com.ruoyi.report.service.open.PrintSendLogService;
8. import lombok.extern.slf4j.Slf4j;
9. import org.springframework.amqp.core.Message;
10. import org.springframework.amqp.rabbit.annotation.RabbitListener;
11. import org.springframework.beans.factory.annotation.Autowired;
12. import org.springframework.stereotype.Component;
14. import java.io.IOException;
16. /**
17. * @ClassName PrintFeedbackConsumer
18. * @Description
19. * @Author Mr.Huang
20. * @Date 2024/4/30 10:23
21. * @Version 1.0
22. **/
23. @Slf4j
24. @Component
25. public class PrintFeedbackConsumer {
27.     @Autowired
28.     private PrintSendLogService printSendLogService;
30.     @RabbitListener(queues = {RabbitMqConfig.print_4pl_dispatch_info_feedback_queue}, containerFactory = "printContainerFactory")
31.     public void receiveMq(Message message, Channel channel) {
32.         try {
33.             String body = new String(message.getBody());
34.             log.info("接受【Print结果推送】RabbitMQ消息："+body);
35.             JSONObject objJson = JSONObject.parseObject(body);
36.             Thread.sleep(1000);
37.             PrintResult printResult = JSONObject.toJavaObject(objJson, PrintResult.class);
38.             printSendLogService.updatePrintSendLog(printResult);
39.         }catch (Exception e){
40.             log.error("",e);
41.             try {
42.                 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
43.             } catch (IOException ex) {
44.                 ex.printStackTrace();
45.             }
46.         }
47.     }
48. }

复制代码

  与单个RabbitMQ消费者利用同等，只是留意要消费的队列和连接工厂不要搞错了
  三、总结 

   配置单个RabbitMQ时不需要关心底层的连接工厂是怎样配置的，当把yaml内容填好它会主动配置连接工厂，只需要把交换机、队列、配置绑定起来即可。 当需要配置多个mq时才需要自己手动配置连接工厂，并不是只能配置两个RabbitMQ，可以按这个格式配置更多个。唯一留意的是不要把这些队列和交换机搞混了即可。

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