SpringBoot 集成多个rabbitmq_springboot 多rabbitmq

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;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

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);
            }
        }
    }
到此就配置好了
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42