Springboot 监听同一个RabbitMQ的多个虚拟主机Virtual-Host_springboot rabbitmq 监听多个 host

场景：需要监听 同一个RabbitMQ下的 不同的虚拟主机下的 队列 来完成数据流转

配置信息application.properties

my.rabbitmq.host=你的ip
my.rabbitmq.port=你的port
my.rabbitmq.username=你的username
my.rabbitmq.password=你的password
my.rabbitmq.first.virtual-host=firstVHost
my.rabbitmq.second.virtual-host=secondVHost
1
2
3
4
5
6

配置类RabbitMqConfig

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 * @author: zdy
 * date: 2020/7/22 16:26
 * version: 1.0.0
 * description: RabbitMQ配置类
 */
@Configuration
public class RabbitMqConfig {

    /**
     * first 连接工厂
     */
    @Bean(name = "firstConnectionFactory")
    public ConnectionFactory firstConnectionFactory(@Value("${my.rabbitmq.host}") String host,
                                                    @Value("${my.rabbitmq.port}") int port,
                                                    @Value("${my.rabbitmq.username}") String username,
                                                    @Value("${my.rabbitmq.password}") String password,
                                                    @Value("${my.rabbitmq.first.virtual-host}") String vHost) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(vHost);
        return connectionFactory;
    }

    /**
     * second 连接工厂
     */
    @Bean(name = "secondConnectionFactory")
    //这个标签是必须的，因为加上后才能有默认值，不会在项目加载时产生冲突
    @Primary
    public ConnectionFactory secondConnectionFactory(@Value("${my.rabbitmq.host}") String host,
                                                     @Value("${my.rabbitmq.port}") int port,
                                                     @Value("${my.rabbitmq.username}") String username,
                                                     @Value("${my.rabbitmq.password}") String password,
                                                     @Value("${my.rabbitmq.second.virtual-host}") String vHost) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(vHost);
        return connectionFactory;
    }


    /**
     *    first生产者
     */
    @Bean(name = "firstRabbitTemplate")
    public RabbitTemplate firstRabbitTemplate(@Qualifier("firstConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    /**
     *   second生产者
     */
    @Bean(name = "secondRabbitTemplate")
    public RabbitTemplate secondRabbitTemplate(@Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    /**
     * first 消费者
     */
    @Bean(name = "firstListenerFactory")
    public SimpleRabbitListenerContainerFactory firstListenerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("firstConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory listenerContainerFactory = new SimpleRabbitListenerContainerFactory();
        //设置手动ack
        listenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        configurer.configure(listenerContainerFactory, connectionFactory);
        return listenerContainerFactory;
    }

    /**
     * second 消费者
     */
    @Bean(name = "secondListenerFactory")
    public SimpleRabbitListenerContainerFactory secondListenerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory listenerContainerFactory = new SimpleRabbitListenerContainerFactory();
        listenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        configurer.configure(listenerContainerFactory, connectionFactory);
        return listenerContainerFactory;
    }

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

监听消息示例

 @RabbitListener(queues = "你的队列名", containerFactory = "firstListenerFactory")
 public void dealDeclareMessage(Channel channel, Message message) throws Exception {
     try {
         //消息类型
         String routingKey = message.getMessageProperties().getReceivedRoutingKey();
         log.info(routingKey);
         //消息内容
         JSONObject jsonObject = JSON.parseObject(new String(message.getBody(), StandardCharsets.UTF_8));
         //处理你的消息
         saveOrUpdate(jsonObject);
     } catch (Exception e) {
         log.info("你的队列名"+"异常信息：" + e.getMessage());
     } finally {
     	 //手动ack
     	 //TODO 也需要异常处理
         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18