SpringBoot 集成多个rabbitmq

比眉伴天荒 2023-02-25 02:15 6阅读 0赞

## SpringBoot 集成多个rabbitmq ##

在项目中经常遇到一个springboot工程要连接多个rabbitmq。如果只连接一个，springboot可以零配置连接rabbitmq，这样不需要做额外的工作。但如果连接多个rabbitmq，就得做一些配置了。  
1 配置多个rabbitmq的连接地址：

spring.rabbitmq.first.host=192.168.10.223  
spring.rabbitmq.first.port=5672  
spring.rabbitmq.first.username=admin  
spring.rabbitmq.first.password=admin

spring.rabbitmq.second.host=192.168.10.224  
spring.rabbitmq.second.port=5672  
spring.rabbitmq.second.username=admin  
spring.rabbitmq.second.password=admin

spring.rabbitmq.third.host=192.168.10.225  
spring.rabbitmq.third.port=5672  
spring.rabbitmq.third.username=admin  
spring.rabbitmq.third.password=admin

2 编写配置类
    
     /*
        定义连接，rabbitMQ
         */
        @Bean(name = "mainConnectionFactory")
        @Primary
        public ConnectionFactory mainConnectionFactory(
                @Value("${spring.rabbitmq.first.host}") String host,
                @Value("${spring.rabbitmq.first.port}") int port,
                @Value("${spring.rabbitmq.first.username}") String username,
                @Value("${spring.rabbitmq.first.password}") String password) {
            return connectionFactory(host, port, username, password);
        }
    
        /*
            定义连接
             */
        @Bean(name = "secondConnectionFactory")
        public ConnectionFactory secondConnectionFactory(
                @Value("${spring.rabbitmq.second.host}") String host,
                @Value("${spring.rabbitmq.second.port}") int port,
                @Value("${spring.rabbitmq.second.username}") String username,
                @Value("${spring.rabbitmq.second.password}") String password) {
            return connectionFactory(host, port, username, password);
        }
    
        /*
         定义连接
          */
        @Bean(name = "thirdConnectionFactory")
        public ConnectionFactory thirdConnectionFactory(
                @Value("${spring.rabbitmq.third.host}") String host,
                @Value("${spring.rabbitmq.third.port}") int port,
                @Value("${spring.rabbitmq.third.username}") String username,
                @Value("${spring.rabbitmq.third.password}") String password) {
            return connectionFactory(host, port, username, password);
        }
    public CachingConnectionFactory connectionFactory(String host, int port, String username, String password) {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
            connectionFactory.setHost(host);
            connectionFactory.setPort(port);
            connectionFactory.setUsername(username);
            connectionFactory.setPassword(password);
            //  connectionFactory.setVirtualHost(virtual_host);
            return connectionFactory;
        }
    
        @Bean(name = "mainRabbitTemplate")
        @Primary
        public RabbitTemplate mainRabbitTemplate(@Qualifier("mainConnectionFactory") ConnectionFactory connectionFactory) {
            RabbitTemplate mainRabbitTemplate = new RabbitTemplate(connectionFactory);
            mainRabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            return mainRabbitTemplate;
        }
    
        @Bean(name = "secondRabbitTemplate")
        public RabbitTemplate secondRabbitTemplate(@Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
            RabbitTemplate secondRabbitTemplate = new RabbitTemplate(connectionFactory);
            return secondRabbitTemplate;
        }
    
        @Bean(name = "thirdRabbitTemplate")
        public RabbitTemplate thridRabbitTemplate(@Qualifier("thirdConnectionFactory") ConnectionFactory connectionFactory) {
            RabbitTemplate thridRabbitTemplate = new RabbitTemplate(connectionFactory);
            return thridRabbitTemplate;
        }
    
        @Bean(name = "mainFactory")
        @Primary
        public SimpleRabbitListenerContainerFactory myFactory(
                SimpleRabbitListenerContainerFactoryConfigurer configurer,
                @Qualifier("mainConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setMessageConverter(new Jackson2JsonMessageConverter());
            configurer.configure(factory, connectionFactory);
            return factory;
        }
    
        @Bean(name = "secondFactory")
        public SimpleRabbitListenerContainerFactory secondFactory(
                SimpleRabbitListenerContainerFactoryConfigurer configurer,
                @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConcurrentConsumers(10);
            factory.setMaxConcurrentConsumers(20);
            factory.setPrefetchCount(1);
            configurer.configure(factory, connectionFactory);
            return factory;
        }
    
        @Bean(name = "thirdFactory")
        public SimpleRabbitListenerContainerFactory thirdFactory(
                SimpleRabbitListenerContainerFactoryConfigurer configurer,
                @Qualifier("thirdConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setMaxConcurrentConsumers(20);
            factory.setConcurrentConsumers(10);
            factory.setPrefetchCount(1);
            configurer.configure(factory, connectionFactory);
            return factory;
        }

3 使用发送消息
    
    
        @Resource(name = "mainRabbitTemplate")
        protected RabbitTemplate rabbitTemplate;
    
    
        public MessageProvider(RabbitTemplate rabbitTemplate) {
            this.rabbitTemplate = rabbitTemplate;
            this.rabbitTemplate.setMandatory(true);
            this.rabbitTemplate.setConfirmCallback(this);
        }
    
        public void sendMessage(RealBusBaseLineDTO realBusBaseLineDTO) {
            if (realBusBaseLineDTO != null) {
                String msg = JSON.toJSONString(realBusBaseLineDTO, SerializerFeature.DisableCircularReferenceDetect);
                rabbitTemplate.convertAndSend(QueueEnum.MESSAGE_REALBUS_QUEUE.getExchange(), QueueEnum.MESSAGE_REALBUS_QUEUE.getRouteKey(), msg);
            } else {
                log.warn("消息内容为空！！！！！");
            }
        }
    
    4 消费消息
    
     @Resource
        private IRealBusDataService realBusDataService;
        @Value("${network.bus.isReadActiveQueue}")
        private Boolean isReadActiveQueue;
    
        @RabbitListener(queues = QueueContent.MESSAGE_REALBUS_DATA_QUEUE_NAME, containerFactory = "secondFactory")
        @RabbitHandler
        public void processRealBusDataQueue(String msg, Channel channel, Message message) throws IOException {
                log.warn("processRealBusDataQueue:{} ", msg);
                try {
                    String bodyMessage = new String(msg);
    				// 逻辑处理
                } catch (Exception e) {
                    log.error("确认消费异常", e);
                }
            }
        }
    到此就配置好了