SpringBoot项目配置多个RabbitMQ解决方案

Posted 2023-03-02 brian0724

背景介绍

目前有一个Spring Boot项目，已经接入一个RabbitMQ Broker，由于业务扩展，需要新增一个RabbitMQ Broker进行消费，由于单个Broker时通过Spring默认配置进行使用，因此需要做出修改

Java客户端开发Demo

ConnectionFactory factory = new ConnectionFactory();
factory.setHost(IP_ADDRESS);
factory.setPort(PORT);
factory.setUsername("root");
factory.setPassword("root123");
factory.setVirtualHost("/");
// 创建链接
Connection connection = factory.newConnection();
// 创建信道，Connection 接口被用来创建一个Channel
Channel channel = connection.createChannel();
// 创建一个type="direct" 持久化、非自动删除的交换机
channel.exchangeDeclare(QUEUE_NAME, "direct", true, false, null);
// 创建一个持久化、非排他的、非自动删除的队列
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
// 发送一条持久化的消息
String message = "hello world!";
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
// 关闭资源
channel.close();
connection.close();

单实例配置

1. pom

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
	<version>2.1.2.RELEASE</version>
</dependency>

2. 增加properties配置

spring.rabbitmq.username=username
spring.rabbitmq.password=password
spring.rabbitmq.host=host
spring.rabbitmq.virtual-host=/
spring.rabbitmq.port=port

3. 增加配置类

@Configuration
public class RabbitMqConfiguration 
	@Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory,MessageConverter messageConverter)
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        return factory;
    

    @Bean
    public MessageConverter messageConverter()
        ObjectMapper objectMapper = new ObjectMapper();
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(objectMapper);
        return jackson2JsonMessageConverter;
    

多实例配置

1. 增加properties配置

# 默认rabbit配置
spring.rabbitmq.username=username
spring.rabbitmq.password=password
spring.rabbitmq.host=host
spring.rabbitmq.virtual-host=/
spring.rabbitmq.port=port
# 第2个MQ配置
spring.rabbitmq.second.host=second.host
spring.rabbitmq.second.username=second.username
spring.rabbitmq.second.password=second.password
spring.rabbitmq.second.port=second.port
spring.rabbitmq.second.virtual-host=/

2. 修改配置类

/**
 * @Description:
 * @Author: admin
 * @Date: 2021/5/11 18:19
 */
@Configuration
public class RabbitMqConfiguration 
    @Primary
    @Bean(name = "rabbitConnectionFactory")
    public ConnectionFactory rabbitConnectionFactory(
            @Value("$spring.rabbitmq.host") String host,
            @Value("$spring.rabbitmq.port") int port,
            @Value("$spring.rabbitmq.username") String username,
            @Value("$spring.rabbitmq.password") String password,
            @Value("$spring.rabbitmq.virtual-host") String virtualHost) 
        return connectionFactory(host, port, username, password, virtualHost);
    

    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(
            @Qualifier("rabbitConnectionFactory") ConnectionFactory connectionFactory,
            MessageConverter messageConverter) 
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        return factory;
    

    /**
     * 定义连接
     * @param host
     * @param port
     * @param username
     * @param password
     * @return
     */
    @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,
            @Value("$spring.rabbitmq.second.virtual-host") String virtualHost) 
        return connectionFactory(host, port, username, password, virtualHost);
    

    /**
     * 声明RabbitTemplate
     * @param connectionFactory
     * @return
     */
    @Bean(name = "secondRabbitTemplate")
    public RabbitTemplate dataadminmqRabbitTemplate(
            @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) 
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate .setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate ;
    
    
    @Bean(name = "secondRabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory secondRabbitListenerContainerFactory(
            @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory,
            MessageConverter messageConverter) 
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        return factory;
    
    
    @Bean
    public MessageConverter messageConverter()
        ObjectMapper objectMapper = new ObjectMapper();
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter(objectMapper);
        return jackson2JsonMessageConverter;
    

    /**
     * 声明连接工厂连接开发服务器
     *
     * @param host
     * @param port
     * @param username
     * @param password
     * @return
     */
    public CachingConnectionFactory connectionFactory(String host, int port, String username,
                                                      String password, String virtualHost) 
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;