Spring Boot 整合 RabbitMQ 多数据测试用例

古城微笑少年丶 2023-10-09 23:39 38阅读 0赞

![d6d8fb263fab46e1b47d9a7edacd6bcb.jpeg][]

大千世界，奇葩需求千千万，为应对开发中可能会遇到到业务场景，需要我们连接多个MQ，因此本篇文章应运而生，希望能得到各位同行大佬指教一二，废话不多说上代码！

1、pom.xml 文件中引入相关依赖；

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
                <version>2.5.5</version>
            </dependency>
    </dependencies>

2、application.yml 定义需要连接的mq信息；

> 注意：小编此处测试使用的是**单节点+集群** 两种连接方式
> 
> 小伙伴们copy时记得根据实际情况调整哦！！！

spring:
      rabbitmq:
        first: #Rabbitmq单节点连接
          host: 172.16.*.* #替换为自己的ip节点
          port: 5672
          username: admin
          password: 123456
          virtual-host: / #无分组默认 /
        second: #Rabbitmq集群连接
          #线上公共
          addresses: 172.16.*.*:5672,172.16.*.*:5672,172.16.*.*:5672
          username: admin
          password: 123456
          virtual-host: **** #分组配置 有需要可以在此处配置
       #新节点可以在下面继续添加

3、连接工厂配置类 创建 config文件 copy一下代码：

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.stereotype.Component;
    
    /**
     * Rabbitmq 多数据源配置
     */
    @Component
    @Configuration
    public class RabbitmqConfig {
    
        private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);
    
        /**************************first***********************/
        @Value("${spring.rabbitmq.first.host}")
        private String host1;
    
        @Value("${spring.rabbitmq.first.port}")
        private Integer port1;
    
        @Value("${spring.rabbitmq.first.username}")
        private String username1;
    
        @Value("${spring.rabbitmq.first.password}")
        private String password1;
    
        @Value("${spring.rabbitmq.first.virtual-host}")
        private String virtualHost1;
        /**************************first**end*****************/
    
        /**************************second***********************/
        //需注意集群和单节点链接区别
        @Value("${spring.rabbitmq.second.addresses}")
        private String addresses2;
    
        private Integer port2;
    
        @Value("${spring.rabbitmq.second.username}")
        private String username2;
    
        @Value("${spring.rabbitmq.second.password}")
        private String password2;
    
        @Value("${spring.rabbitmq.second.virtual-host}")
        private String virtualHost2;
    
        /**************************second**end*****************/
    
    
        /**************************自定义***********************/
        //TODO 后续需添加其他数据源 复制以上代码 即可！
        /**************************自定义**end*****************/
    
        /**
         * 通用连接工厂工具类
         * @param isColony 是否为集群连接
         * @param host ip
         * @param port 端口
         * @param username 用户名
         * @param password 密码
         * @param virtualHost 分组名
         * @return
         */
        public ConnectionFactory connectionFactory(boolean isColony,String host, Integer port, String username, String password, String virtualHost) {
            //CachingConnectionFactory 是一个连接池，需要显式的关闭 ,配置集群
            log.info("【{}】host:【{}】port:【{}】username:【{}】password:【{}】virtualHost:【{}】","RabbitmqConfig",host,port,username,password,virtualHost);
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            if (isColony) {
                connectionFactory.setAddresses(host);
            }else {
                connectionFactory.setHost(host);
                connectionFactory.setPort(port);
            }
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
            return connectionFactory;
        }
    
        @Bean(name = "firstConnectionFactory")
        @Primary
        public ConnectionFactory firstConnectionFactory() {
            return connectionFactory(false,host1,port1,username1,password1,virtualHost1);
        }
    
        @Bean(name = "firstRabbitTemplate")
        @Primary
        public RabbitTemplate firstRabbitTemplate() {
            RabbitTemplate firstRabbitTemplate = new RabbitTemplate(firstConnectionFactory());
            return firstRabbitTemplate;
        }
    
        @Bean(name = "firstFactory")
        public SimpleRabbitListenerContainerFactory firstFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("firstConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            configurer.configure(factory, connectionFactory);
            return factory;
        }
    
    
        @Bean(name = "secondConnectionFactory")
        public ConnectionFactory secondConnectionFactory() {
            return connectionFactory(true,addresses2,null,username2,password2,virtualHost2);
        }
    
        @Bean(name = "secondRabbitTemplate")
        public RabbitTemplate secondRabbitTemplate() {
            RabbitTemplate firstRabbitTemplate = new RabbitTemplate(secondConnectionFactory());
            return firstRabbitTemplate;
        }
    
        @Bean(name = "secondFactory")
        public SimpleRabbitListenerContainerFactory secondFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            configurer.configure(factory, connectionFactory);
            return factory;
        }
    
    }

4、测试用例！ 小主！别急，已经到了最后一步，再坚持一下...

import com.alibaba.fastjson.JSONObject;
    import com.rabbitmq.client.Channel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.AmqpHeaders;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.messaging.handler.annotation.Header;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     *测试控制类
     */
    @RestController
    public class TestController {
    
        private static final Logger log = LoggerFactory.getLogger(TestController.class);
    
        @Autowired
        @Qualifier("firstRabbitTemplate")
        private RabbitTemplate firstRabbitTemplate;
    
        @Autowired
        @Qualifier("secondRabbitTemplate")
        private RabbitTemplate secondRabbitTemplate;
    
        /**
         * 提交测试数据至消息队列
         */
        @GetMapping("sendData")
        public void sendData(){
            String info = "这是一条测试数据！";
            firstRabbitTemplate.convertAndSend("test_info_queue", JSONObject.toJSONString(info + "【firstRabbitTemplate】"));
            log.info("数据源:【{}】消息提交成功！","first");
            secondRabbitTemplate.convertAndSend("test_info_queue",JSONObject.toJSONString(info) + "【secondRabbitTemplate】");
            log.info("数据源:【{}】消息提交成功！","second");
        }
    
        /**
         * first 数据源
         * @param message 传输的对象
         * @param channel 处理的管道
         * @param tag
         */
        @RabbitListener(queues = "test_info_queue",containerFactory = "firstFactory")
        public void onMessageReceiver1(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag){
            try {
                log.info("数据源:【{}】队列:【{}】消费成功！msg:【{}】","first","test_info_queue",message);
            } catch (Exception e) {
              log.error("数据源:【{}】队列:【{}】消费异常！msg:【{}】","first","test_info_queue",e.getMessage());
            }
        }
    
        /**
         * second 数据源
         * @param message 传输的对象
         * @param channel 处理的管道
         * @param tag
         */
        @RabbitListener(queues = "test_info_queue",containerFactory = "secondFactory")
        public void onMessageReceiver2(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag){
            try {
                log.info("数据源:【{}】队列:【{}】消费成功！msg:【{}】","second","test_info_queue",message);
            } catch (Exception e) {
                log.error("数据源:【{}】队列:【{}】消费异常！msg:【{}】","second","test_info_queue",e.getMessage());
            }
        }
    
    
    }

5、运行程序，测试是否成功！

出现以下结果，恭喜小主完成调试，请继续在代码的汪洋大海中，乘风破浪 ~~~

o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring DispatcherServlet 'dispatcherServlet'
    o.s.web.servlet.DispatcherServlet  : Initializing Servlet 'dispatcherServlet'
    o.s.web.servlet.DispatcherServlet  : Completed initialization in 1 ms
    c.k.m.controller.TestController    : 数据源:【first】消息提交成功！
    c.k.m.controller.TestController    : 数据源:【first】队列:【test_info_queue】消费成功！msg:【"###firstRabbitTemplate###这是一条测试数据！"】
    c.k.m.controller.TestController    : 数据源:【second】消息提交成功！
    c.k.m.controller.TestController    : 数据源:【second】队列:【test_info_queue】消费成功！msg:【"###secondRabbitTemplate###这是一条测试数据！"】

此篇，至此结束！

如若这边文章对你有所帮助，动动发财小手，帮小编给个赞！

众人肯定是博主创作前进动力的源泉，以后会更加努力创作新的文章

[d6d8fb263fab46e1b47d9a7edacd6bcb.jpeg]: https://img-blog.csdnimg.cn/d6d8fb263fab46e1b47d9a7edacd6bcb.jpeg