SpringBoot 集成多个rabbitmq_sebeefe_springboot 多个rabbitmq

未知 02-07 7179

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); } } } 到此就配置好了