RabbitMQ | SpringBoot集成多个RabbitMQ_项目中集成不同的rabbitmq 服务

一、前言

在开发过程中需要对接多个系统的RabbitMQ 连接，所以需要自己创建RabbitMQ连接, 以下demo 项目中用两个RabbitMQ 做了演示, 项目中包含了 五种模式的 消息发送以及消息接收

代码有点多 如果看不下去可以直接将项目拉下来参考

RabbitMQ 简介: https://blog.csdn.net/qq825478739/article/details/125932427

Demo项目gitee地址: https://gitee.com/yijiecaomin/springboot-rabbitmq-demo.git

二、 Demo项目

1. 项目整体

2. Pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.example</groupId>
        <artifactId>rabbitmq-demo-parent</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>

    <groupId>com.rabbit.many</groupId>
    <artifactId>rabbitmq-many</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-many</name>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

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

3. application.properties

####自定义默认MQ 连接
spring.rabbitmq.default-host=192.168.8.101
spring.rabbitmq.default-port=5672
spring.rabbitmq.default-username=user1
spring.rabbitmq.default-password=user1
## 自定义MQ2 配置
spring.rabbitmq.two-host=192.168.8.105
spring.rabbitmq.two-port=5672
spring.rabbitmq.two-username=zhkj
spring.rabbitmq.two-password=zhkj
1
2
3
4
5
6
7
8
9
10

4. MQProperties 自定义配置属性

@Data
@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class MQProperties {

    /**
     * 默认MQip
     */
    private String defaultHost;

    /**
     * 默认MQ端口
     */
    private Integer defaultPort;

    /**
     * 默认MQ用户名
     */
    private String defaultUsername;

    /**
     * 默认MQ密码
     */
    private String defaultPassword;

    /**
     * MQ2 ip
     */
    private String twoHost;

    /**
     * MQ2 端口
     */
    private Integer twoPort;

    /**
     * MQ2 用户名
     */
    private String twoUsername;

    /**
     * MQ2 密码
     */
    private String twoPassword;

}
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

5. MQFactoryConfig 配置 这里是核心哦

/**
 * MQ 工厂配置类
 *
 * @Author su
 * @Date 2022/7/22 11:36
 */
@Configuration
public class MQFactoryConfig {

    @Autowired
    private MQProperties rabbitMqProperties;

    //线程数量:根据实际的服务器性能进行配置即可
    public static final int DEFAULT_CONCURRENT = 10;
    //最大线程数
    public static final int MAX_CONCURRENT = 48;

    /**
     * 默认连接工厂
     *
     * @return
     */
    @Primary
    @Bean(name = "defaultConnectionFactory")
    public ConnectionFactory defaultConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(rabbitMqProperties.getDefaultHost());
        connectionFactory.setPort(rabbitMqProperties.getDefaultPort());
        connectionFactory.setUsername(rabbitMqProperties.getDefaultUsername());
        connectionFactory.setPassword(rabbitMqProperties.getDefaultPassword());
        return connectionFactory;
    }


    /**
     * 默认容器工厂
     *
     * @param configurer
     * @param connectionFactory
     * @return
     */
    @Primary
    @Bean(name = "defaultFactory")
    public SimpleRabbitListenerContainerFactory defaultFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("defaultConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // 指定要创建的并发消费者数量   默认值为1
        factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
        // 设置消费者数量的上限
        factory.setMaxConcurrentConsumers(MAX_CONCURRENT);
        // 自动应答
        // factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        //消息转换器
        //factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //启动事务模式
        //factory.setChannelTransacted(true);
        //设置重试
        factory.setAdviceChain(createRetry());
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    /**
     * 创建重试
     *
     * @return
     */
    private RetryOperationsInterceptor createRetry() {
        RetryTemplate retryTemplate = new RetryTemplate();

        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(3000); // 重试间隔
        backOffPolicy.setMaxInterval(6000); // 重试最大间隔
        //backOffPolicy.setMultiplier(2000);// 重试间隔乘法策略
        retryTemplate.setBackOffPolicy(backOffPolicy);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);
        return RetryInterceptorBuilder.stateless().retryOperations(retryTemplate).recoverer(new RejectAndDontRequeueRecoverer()).build();
    }

    /**
     * 默认RabbitTemplate操作类
     *
     * @param connectionFactory
     * @return
     */
    @Primary
    @Bean(name = "defaultTemplate")
    public RabbitTemplate defaultTemplate(@Qualifier("defaultConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    @Primary//这个注解一定要加 Rabbitmq需要一个默认的 工厂
    @Bean(name = "defaultRabbitAdmin")
    public RabbitAdmin defaultRabbitAdmin(@Qualifier("defaultConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * MQ 2 连接工厂
     *
     * @return
     */
    @Bean(name = "twoConnectionFactory")
    public ConnectionFactory MergeConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(rabbitMqProperties.getTwoHost());
        connectionFactory.setPort(rabbitMqProperties.getTwoPort());
        connectionFactory.setUsername(rabbitMqProperties.getTwoUsername());
        connectionFactory.setPassword(rabbitMqProperties.getTwoPassword());
        return connectionFactory;
    }

    /**
     * MQ 2 容器工厂
     *
     * @param configurer
     * @param connectionFactory
     * @return
     */
    @Bean(name = "twoFactory")
    public SimpleRabbitListenerContainerFactory LocalFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, @Qualifier("twoConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置线程数量
        factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
        factory.setMaxConcurrentConsumers(MAX_CONCURRENT);
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    /**
     * MQ 2 操作类
     *
     * @param connectionFactory
     * @return
     */
    @Bean(name = "twoTemplate")
    public RabbitTemplate TwoTemplate(@Qualifier("twoConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    @Bean(name = "twoRabbitAdmin")
    public RabbitAdmin twoRabbitAdmin(@Qualifier("twoConnectionFactory") ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
}
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

6. 消息生产者代码

6.1 简单模式 /工作模式 发布者

创建队列

/**
 * 点对点 模式 配置
 *  简单模式     work工作模式
 *
 * @Author su
 * @Date 2022/7/21 15:50
 */
@Configuration
public class SimpleWorkRabbitMQConfig {

    /**
     * 简单队列  一个发布者 ->队列 -> 一个消费者
     *
     *  一个消费者消费队列中消息
     *
     * @return
     */
    @Bean
    public Queue simpleQueue(){
        return new Queue("simple.demo",true);
    }

    /**
     * 工作队列 : 一个发布者 -> 队列 -> 多个消费者
     * 工作模式与简单模式 只是消费者的数量问题 比如说将 队列当做一个快递站  消费者是快递员 将消息分发给大家来处理
     *
     * @return
     */
    @Bean
    public Queue workQueue(){
        return new Queue("work.demo",true);
    }

}
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

创建消息生产服务

/**
 * 简单模式 / 工作模式 发送者
 *
 * @Author su
 * @Date 2022/7/21 15:55
 */
@Component
public class SimpleWorkMQProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 另个RabbitTemplate  需要用这个发送 就使用这个
     */
    @Resource(name = "twoTemplate")
    private RabbitTemplate twoTemplate;


    /**
     * 发送简单 MQ
     */
    public void sendSimpleMq() {
        for (int i = 1; i <= 3; i++) {
            rabbitTemplate.convertAndSend("simple.demo", "发送简单消息内容: 简单简单简单 :" + i);
        }

    }

    /**
     * 发送工作 MQ
     */
    public void sendWorkMq() {
        for (int i = 1; i <= 10; i++) {
            rabbitTemplate.convertAndSend("work.demo", "发送工作消息内容: 工作工作简单 :" + i);
        }
    }
}
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

6.2 Fanout 广播模式 发布者

创建交换机: 发布订阅模式 发布者不需要创建队列

/**
 * 广播模式创建交换机
 *
 * @Author su
 * @Date 2022/7/21 17:43
 */
@Configuration
public class FanoutRabbitMQConfig {

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout.exchange.demo", true, false);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

创建消息生产服务

/**
 * 广播模式发送消息
 *
 * @Author su
 * @Date 2022/7/21 17:47
 */
@Component
public class FanoutMQProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "twoTemplate")
    private RabbitTemplate twoTemplate;

    public void sendFanoutMq() {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend("fanout.exchange.demo", null, "Fanout 消息来了 :" + i);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

6.3 Routing 路由模式 发布者

创建交换机: 发布订阅模式 发布者不需要创建队列

/**
 * 路由 配置类
 * 1.创建交换机
 * 2.绑定路由
 *
 * @Author su
 * @Date 2022/7/21 18:03
 */
@Configuration
public class RoutingRabbitMQConfig {

    /**
     * 交换机
     *
     * @return
     */
    @Bean
    public DirectExchange getDirectExchange() {
        //return new DirectExchange("routing.rabbit.demo",true,false);
        return ExchangeBuilder.directExchange("routing.exchange.demo").durable(true).build();
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

创建消息生产服务

/**
 * 广播模式发送消息
 *
 * @Author su
 * @Date 2022/7/21 17:47
 */
@Component
public class RoutingMQProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "twoTemplate")
    private RabbitTemplate twoTemplate;

    public void sendRoutingMq() {
        for (int i = 0; i < 5; i++) {
            if (i % 2 == 0) {
                rabbitTemplate.convertAndSend("routing.exchange.demo", "routing.msg.2", "Routing 消息来了 :" + i);
            } else {
                rabbitTemplate.convertAndSend("routing.exchange.demo", "routing.msg.1", "Routing 消息来了 :" + i);
            }
        }
    }
}

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

6.4 Topic 主题模式 发布者

创建交换机: 发布订阅模式 发布者不需要创建队列

/**
 * topic 模式
 *
 * @Author su
 * @Date 2022/7/22 10:01
 */
@Configuration
public class TopicRabbitMQConfig {

    /**
     * topic 交换机
     *
     * @return
     */
    @Bean
    public TopicExchange getTopicExchange() {
        return ExchangeBuilder.topicExchange("topic.exchange.demo").durable(true).build();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

创建消息生产服务

/**
 * 广播模式发送消息
 *
 * @Author su
 * @Date 2022/7/21 17:47
 */
@Component
public class TopicMQProducer {

    @Resource(name = "defaultTemplate")
    private RabbitTemplate rabbitTemplate;

    @Resource(name = "twoTemplate")
    private RabbitTemplate twoTemplate;

    public void sendTopicMq() {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend("topic.exchange.demo", "topic.demo", "Topic 消息来了 :" + i);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

7.消费者代码

7.1 简单模式 /工作模式 消费者

/**
 * 工作模式 /简单模式 消费者
 * <p>
 * 需要注意 :消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
 *
 * @Author su
 * @Date 2022/7/21 16:06
 */
@Component
public class SimpleWorkMQConsumer {

    /**
     * 简单模式 消费者
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(queues = "simple.demo",containerFactory = "defaultFactory")
    public void simpleMsg(String msg) {
        System.out.println(new SimpleDateFormat("hh:mm:ss").format(new Date()) +" 简单模式 接收到消息 :" + msg);
          //int i = 1 / 0;
        //消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
    }

    /**
     * 工作模式 消费者1号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(queues = "work.demo",containerFactory = "defaultFactory")
    public void workMsg1(String msg) {
        System.out.println("工作模式 1号工作者接收到消息 :" + msg);
    }

    /**
     * 工作模式 消费者2号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(queues = "work.demo",containerFactory = "defaultFactory")
    public void workMsg2(String msg) {
        System.out.println("工作模式 2号工作者接收到消息 :" + msg);
    }
}
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

7.2 Fanout 广播模式 消费者

/**
 * 广播模式 消费者
 *
 * @Author su
 * @Date 2022/7/21 17:52
 */
@Component
public class FanoutMQConsumer {

    /**
     * 广播模式 消费者1号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "fanout.queue.demo"), exchange = @Exchange(value = "fanout.exchange.demo", type = ExchangeTypes.FANOUT)),containerFactory = "defaultFactory")
    public void fanoutMsg(String msg) {
        System.out.println("广播模式 1号 接收到消息 :" + msg);
    }

    /**
     * 广播模式 消费者2号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "fanout.queue.demo2"), exchange = @Exchange(value = "fanout.exchange.demo", type = ExchangeTypes.FANOUT)),containerFactory = "defaultFactory")
    public void fanoutMsg2(String msg) {
        System.out.println("广播模式 2号 接收到消息 :" + msg);
    }
}
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

7.3 Routing 路由模式 消费者

/**
 * 路由模式 消费者
 *
 * @Author su
 * @Date 2022/7/21 17:52
 */
@Component
public class RoutingMQConsumer {

    /**
     * 广播模式 消费者1号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "routing.queue.demo"), exchange = @Exchange(value = "routing.exchange.demo", type = ExchangeTypes.DIRECT),key = "routing.msg.1"),containerFactory = "defaultFactory")
    public void fanoutMsg(String msg) {
        System.out.println("路由模式 1号 接收到消息 :" + msg);
    }

    /**
     * 广播模式 消费者2号
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "routing.queue.demo2"), exchange = @Exchange(value = "routing.exchange.demo", type = ExchangeTypes.DIRECT),key = "routing.msg.2"),containerFactory = "defaultFactory")
    public void fanoutMsg2(String msg) {
        System.out.println("路由模式 2号 接收到消息 :" + msg);
    }
}

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

7.4 Topic 主题模式 消费者

/**
 * 工作模式 /简单模式 消费者  containerFactory指向那个工厂
 * <p>
 * 需要注意 :消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
 *
 * @Author su
 * @Date 2022/7/21 16:06
 */
@Component
public class TopicMQConsumer {

    /**
     * 简单模式 消费者  TODO 可以收到 #多个单词模糊匹配   发送者的路由是:topic.demo
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue("topic.queue.demo1"), exchange = @Exchange(name = "topic.exchange.demo", type = ExchangeTypes.TOPIC), key = "topic.#"), containerFactory = "defaultFactory")
    public void simpleMsg1(String msg) {
        System.out.println("topic模式 topic.# 接收到消息 :" + msg);
        //int i = 1 / 0;
        //消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
    }

    /**
     * 简单模式 消费者  //TODO 可以收到 单个单词 *模糊匹配   发送者的路由是:topic.demo
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue("topic.queue.demo2"), exchange = @Exchange(name = "topic.exchange.demo", type = ExchangeTypes.TOPIC), key = "topic.*"), containerFactory = "defaultFactory")
    public void simpleMsg2(String msg) {
        System.out.println("topic模式 topic.* 接收到消息 :" + msg);
        //int i = 1 / 0;
        //消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
    }

    /**
     * 简单模式 消费者  //TODO 这个路由key为 topic 是收不到消息的  发送者的路由是:topic.demo
     *
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(value = @Queue("topic.queue.demo3"), exchange = @Exchange(name = "topic.exchange.demo", type = ExchangeTypes.TOPIC), key = "topic"), containerFactory = "defaultFactory")
    public void simpleMsg3(String msg) {
        System.out.println("topic模式 topic 接收到消息 :" + msg);
        //int i = 1 / 0;
        //消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理
    }
}
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

8. 测试验证代码

这里写了个 SpringBoot 启动运行的方法 测试验证

/**
 * 启动 发送MQ 的类
 *
 * @Author su
 * @Date 2022/7/21 16:14
 */
@Component
public class RabbitMqDemoStarter implements ApplicationRunner {

    @Autowired
    private SimpleWorkMQProducer simpleWorkMQProducer;
    @Autowired
    private FanoutMQProducer fanoutMQProducer;
    @Autowired
    private RoutingMQProducer routingMQProducer;
    @Autowired
    private TopicMQProducer topicMQProducer;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        System.out.println("Simple---------------------------------");
        simpleWorkMQProducer.sendSimpleMq();

        Thread.sleep(5000);
        System.out.println("Work---------------------------------");
        simpleWorkMQProducer.sendWorkMq();

        Thread.sleep(5000);
        System.out.println("Fanout---------------------------------");
        fanoutMQProducer.sendFanoutMq();

        Thread.sleep(5000);
        System.out.println("Routing---------------------------------");
        routingMQProducer.sendRoutingMq();

        Thread.sleep(5000);
        System.out.println("Topic---------------------------------");
        topicMQProducer.sendTopicMq();
    }
}

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

三、启动验证

四、重点注意总结

有总结的不对的地方/或者问题 请指正, 我在努力中

TODO 需要重点注意

1. 消费者消费消息的时候，发生异常情况，导致消息未确认，该消息会被重复消费(默认没有重复次数，即无限循环消费)，但可以通过设置重试次数以及达到重试次数之后的消息处理

2. 当用注解 @RabbitListener 指向非默认数据源@Primary containerFactory = “其他数据源” 注解不会自动创建新的队列,会报找不到 队列的错误,暂时我只知道手动创建的解决方案 有其他的求评论 告知一下
   手动创建方案: 请查看 OtherQueueMqConfig 这个配置手动创建
   不会自动创建的例子:@RabbitListener(bindings = @QueueBinding(value = @Queue(“新队列名”), exchange = @Exchange(name = “指定交换机”, type = ExchangeTypes.TOPIC), key = “路由”), containerFactory = “其他数据源”)

3. 在创建工厂的时候一定要使用 @Primary 这个注解，RabbitMQ会根据类型注入所以需要一个默认的主数据源 不然会报错: required a single bean, but 2 were found

OtherQueueMqConfig

/**
 * 其他数据源队列创建配置
 *
 * @Author su
 * @Date 2022/7/28 11:52
 */
@Configuration
public class OtherQueueMqConfig {

    @Resource()
    private RabbitAdmin rabbitAdmin;

    @Bean
    void EcuOperationLogQueueDirect() {
        //新队列
        Queue queue = QueueBuilder.durable("new.test").build();
        rabbitAdmin.declareQueue(queue);
        DirectExchange directExchange = new DirectExchange("testExchange",true,false);
        rabbitAdmin.declareExchange(directExchange);
        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with("testRouting"));
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23