SpringBoot整合多個RabbitMQ

時間 2020-08-07

標籤 springboot 整合多個 rabbitmq 欄目 Spring 简体版

原文原文鏈接

1、背景

最近項目中須要用到了RabbitMQ來監聽消息隊列，監聽的消息隊列的虛擬主機(virtualHost)和隊列名(queueName)是不一致的，可是接收到的消息格式相同的。並且可能還存在程序不停機的狀況下，動態的增長新的隊列(queue)的監聽，所以就須要咱們本身在程序中實現一種方法實現動態配置RabbitMQ。java

2、需求

咱們有2個RabbitMQ的配置，在程序啓動的時候，動態的配置好這2個RabbitMQ，實現消息的監聽。git

RabbitMQ的配置信息web

host	port	username	password	virtualHost	queueName
47.101.130.164	5672	rabbit-multi-01	rabbit-multi-01	/rabbit-multi-01	queue-rabbit-multi-01
47.101.130.164	5672	rabbit-multi-02	rabbit-multi-02	/rabbit-multi-02	queue-rabbit-multi-02

3、實現思路

一、動態配置RabbitMQ

包括 ConnectionFactory,RabbitAdmin,RabbitTemplate,SimpleMessageListenerContainer等spring

二、將上方配置好的Bean注入到Spring容器中，以後可能須要用到

向Spring容器中注入Bean的方法springboot

DefaultListableBeanFactory#registerSingleton

DefaultListableBeanFactory#registerBeanDefinition

4、實現步驟

一、引入maven依賴

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

二、建立RabbitProperties用來表示RabbitMQ的配置信息

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class RabbitProperties {
    private String host;
    private Integer port;
    private String username;
    private String password;
    private String virtualHost;
    private String queueName;
}

三、配置RabbitMQ

配置 ConnectionFactory,RabbitAdmin,RabbitTemplate,SimpleMessageListenerContainer等，並動態注入到Spring容器中併發

@Configuration
@RequiredArgsConstructor
@Slf4j
public class MultiRabbitMqConfig {

    private final DefaultListableBeanFactory defaultListableBeanFactory;

    private static Map<String, RabbitProperties> multiMqPropertiesMap = new HashMap<String, RabbitProperties>() {
        {
            put("first", RabbitProperties.builder()
                    .host("47.101.130.164")
                    .port(5672)
                    .username("rabbit-multi-01")
                    .password("rabbit-multi-01")
                    .virtualHost("/rabbit-multi-01")
                    .queueName("queue-rabbit-multi-01").build());
            put("second", RabbitProperties.builder()
                    .host("47.101.130.164")
                    .port(5672)
                    .username("rabbit-multi-02")
                    .password("rabbit-multi-02")
                    .virtualHost("/rabbit-multi-02")
                    .queueName("queue-rabbit-multi-02").build());
        }
    };

    @PostConstruct
    public void initRabbitmq() {
        multiMqPropertiesMap.forEach((key, rabbitProperties) -> {

            AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(CachingConnectionFactory.class)
                    .addPropertyValue("cacheMode", CachingConnectionFactory.CacheMode.CHANNEL)
                    .addPropertyValue("host", rabbitProperties.getHost())
                    .addPropertyValue("port", rabbitProperties.getPort())
                    .addPropertyValue("username", rabbitProperties.getUsername())
                    .addPropertyValue("password", rabbitProperties.getPassword())
                    .addPropertyValue("virtualHost", rabbitProperties.getVirtualHost())
                    .getBeanDefinition();
            String connectionFactoryName = String.format("%s%s", key, "ConnectionFactory");
            defaultListableBeanFactory.registerBeanDefinition(connectionFactoryName, beanDefinition);
            CachingConnectionFactory connectionFactory = defaultListableBeanFactory.getBean(connectionFactoryName, CachingConnectionFactory.class);

            String rabbitAdminName = String.format("%s%s", key, "RabbitAdmin");
            AbstractBeanDefinition rabbitAdminBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(RabbitAdmin.class)
                    .addConstructorArgValue(connectionFactory)
                    .addPropertyValue("autoStartup", true)
                    .getBeanDefinition();
            defaultListableBeanFactory.registerBeanDefinition(rabbitAdminName, rabbitAdminBeanDefinition);
            RabbitAdmin rabbitAdmin = defaultListableBeanFactory.getBean(rabbitAdminName, RabbitAdmin.class);
            log.info("rabbitAdmin:[{}]", rabbitAdmin);

            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            defaultListableBeanFactory.registerSingleton(String.format("%s%s", key, "RabbitTemplate"), rabbitTemplate);

            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
            // 設置監聽的隊列
            simpleMessageListenerContainer.setQueueNames(rabbitProperties.getQueueName());
            // 指定要建立的併發使用者的數量,默認值是1,當併發高時能夠增長這個的數值，同時下方max的數值也要增長
            simpleMessageListenerContainer.setConcurrentConsumers(3);
            // 最大的併發消費者
            simpleMessageListenerContainer.setMaxConcurrentConsumers(10);
            // 設置是否重回隊列
            simpleMessageListenerContainer.setDefaultRequeueRejected(false);
            // 設置簽收模式
            simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            // 設置非獨佔模式
            simpleMessageListenerContainer.setExclusive(false);
            // 設置consumer未被 ack 的消息個數
            simpleMessageListenerContainer.setPrefetchCount(1);
            // 設置消息監聽器
            simpleMessageListenerContainer.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
                try {
                    log.info("============> Thread:[{}] 接收到消息:[{}] ", Thread.currentThread().getName(), new String(message.getBody()));
                    log.info("====>connection:[{}]", channel.getConnection());
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    // 發生異常此處須要捕獲到
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            });
            defaultListableBeanFactory.registerSingleton(String.format("%s%s", key, "SimpleMessageListenerContainer"), simpleMessageListenerContainer);
        });
        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            RabbitTemplate firstRabbitTemplate = (RabbitTemplate) defaultListableBeanFactory.getBean("firstRabbitTemplate");
            firstRabbitTemplate.convertAndSend("exchange-rabbit-multi-01", "", "first queue message");
            log.info("over...");
        }).start();
    }
}