Spring Boot 整合 RabbitMQ 详解

前言：
在消息中间件领域中 RabbitMQ 也是一种非经常见的消息中间件了，本篇简单分享一下 Spring Boot 项目集成 RabbitMQ 的过程。
RabbitMQ 系列文章传送门
RabbitMQ 的先容及焦点概念解说
@RabbitListener 注解详解
Spring Boot 集成 RabbitMQ 可以分为三大步，如下：

• 在 proerties 大概 yml 文件中添加 RabbitMQ 设置。
  
• 项目 pom.xml 文件中引入 spring-boot-starter-amqp 依靠。
  
• 注入 RabbitTemplate 开始利用 RabbitMQ ，其实这步以及算是利用了，不能算作集成了，但是集成了总归是要利用的，我把这里也算作一步了。
  

在 proerties 大概 yml 文件中添加 RabbitMQ 设置如下：

1. spring.rabbitmq.host= xxx.xxx.xxx
2. spring.rabbitmq.port= 5672
3. spring.rabbitmq.username= admin
4. spring.rabbitmq.password= admin
5. spring.rabbitmq.virtual-host = /study

复制代码

项目 pom.xml 文件中引入 spring-boot-starter-amqp 依靠如下：

1. <dependency>
2.         <groupId>org.springframework.boot</groupId>
3.         <artifactId>spring-boot-starter-amqp</artifactId>
4.         <version>2.4.5</version>
5. </dependency>

复制代码

RabbitMQ 利用

前文我们在分享 RabbitMQ 焦点概念的时间，我们知道了 RabbitMQ 有六种互换机类型，下面我们就针对六种互换机来分享 RabbitMQ 的利用。
Direct Exchange（直连互换机）
直连互换机是 RoutingKey 完全匹配模式，也就是我们常说的点对点模式，消息会传送给 RoutingKey 完全匹配的队列。
直连互换机 Direct Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.Binding;
4. import org.springframework.amqp.core.BindingBuilder;
5. import org.springframework.amqp.core.DirectExchange;
6. import org.springframework.amqp.core.Queue;
7. import org.springframework.context.annotation.Bean;
8. import org.springframework.context.annotation.Configuration;
11. @Configuration
12. public class RabbitDirectConfig {
14.     //注入队列
15.     @Bean
16.     public Queue queue() {
17.         return new Queue("direct-buget-queue");
18.     }
20.     //注入交换机
21.     @Bean
22.     public DirectExchange directExchange() {
23.         //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
24.         return new DirectExchange("direct-buget-exchange", true, true);
25.     }
27.     //绑定队列和交换机
28.     @Bean
29.     public Binding binding() {
30.         return BindingBuilder.bind(queue()).to(directExchange()).with("direct-buget-exchange-routing-key");
31.     }
33. }

复制代码

直连互换机 Direct Exchange 消息生产代码如下：

1. //direct 模式消息发送
2. public void sendDirectMessage(String message) {
3.         rabbitTemplate.convertAndSend("direct-buget-exchange", "direct-buget-exchange-routing-key", message);
4. }

复制代码

直连互换机 Direct Exchange 消息消费代码如下：

1. //direct 直连模式消费端
2. @RabbitListener(queues = "direct-buget-queue")
3. public void directConsumer(String message) {
4.         System.out.println("direct 消息消费成功,message内容为:" + message);
5. }

复制代码

直连互换机 Direct Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-direct-buget-message")
2. private String sendDirectBugetMessage(@RequestParam String message) {
3.         myRabbitProducer.sendDirectMessage(message);
4.         return "OK";
5. }

复制代码

直连互换机 Direct Exchange 消息测试（触发消息生产及消费）效果如下：

1. direct 消息消费成功,message内容为:hello message

复制代码

直连互换机，一对一模式，效果符合预期。
Topic Exchange（主题互换机）
主题互换机支持路由模糊匹配，可以利用星号和井号（#）作为通配符进行匹配，其中 ”*“ 可以代替一个单词，（#） 可以代替恣意个单词。
主题互换机 Topic Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
8. @Configuration
9. public class RabbitTopicConfig {
11.     //注入队列
12.     @Bean("topicQueue")
13.     public Queue queue1() {
14.         return new Queue("topic-buget-queue");
15.     }
17.     //注入队列
18.     @Bean("topicQueue2")
19.     public Queue queue2() {
20.         return new Queue("topic-buget-queue2");
21.     }
23.     //注入交换机
24.     @Bean
25.     public TopicExchange topicExchange() {
26.         //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
27.         return new TopicExchange("topic-buget-exchange", true, true);
28.     }
30.     //绑定队列和交换机
31.     @Bean("topicBinding")
32.     public Binding binding() {
33.         return BindingBuilder.bind(queue1()).to(topicExchange()).with("topic.buget.exchange.routing.key.*");
34.     }
35.     //绑定队列和交换机
36.     @Bean("topicBinding2")
37.     public Binding binding2() {
38.         return BindingBuilder.bind(queue2()).to(topicExchange()).with("topic.buget.exchange.routing.key.#");
39.     }
41. }

复制代码

主题互换机 Topic Exchange 消息生产代码如下：

1. //topic 模式消息发送
2. public void sendTopicMessage(String message) {
3.         rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1", message);
4. }
6. //topic 模式消息发送
7. public void sendTopicMessage2(String message) {
8.         rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1.1", message);
9. }

复制代码

主题互换机 Topic Exchange 消息消费代码如下：

1. //topic 模式消费端
2. @RabbitListener(queues = "topic-buget-queue")
3. public void topicConsumer(String message) {
4.         System.out.println("topic topic-buget-queue 消息消费成功,message内容为:" + message);
5. }
7. //topic 模式消费端
8. @RabbitListener(queues = "topic-buget-queue2")
9. public void topicConsumer2(String message) {
10.         System.out.println("topic topic-buget-queue2 消息消费成功,message内容为:" + message);
11. }

复制代码

主题互换机 Topic Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-topic-buget-message")
2. private String sendTopicBugetMessage(@RequestParam String message) {
3.         myRabbitProducer.sendTopicMessage(message);
4.         myRabbitProducer.sendTopicMessage2(message);
5.         return "OK";
6. }

复制代码

主题互换机 Topic Exchange 消息测试（触发消息生产及消费）效果如下：

1. topic topic-buget-queue 消息消费成功,message内容为:hello topic
2. topic topic-buget-queue2 消息消费成功,message内容为:hello topic
3. topic topic-buget-queue2 消息消费成功,message内容为:hello topic

复制代码

主题互换机 Topic Exchange 我们声明了两个 topic.buget.exchange.routing.key.* 和 topic.buget.exchange.routing.key.#，其中星号可以代替一个单词，（#） 可以代替恣意个单词，因此 topic-buget-queue 只能匹配到 topic.buget.exchange.routing.key.1 的消息，topic topic-buget-queue2 则可以匹配到两个路由的消息，效果符合预期。
Fanout Exchange（扇形互换机）
扇形互换机，一个互换器可以绑定多个队列，只要互换机接收到消息就会发送给所有和它绑定的队列，不再进行 RoutingKey 判定，也就是我们常说的发布订阅模式。
扇形互换机 Fanout Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. @Configuration
8. public class RabbitFanoutConfig {
10.     //注入队列
11.     @Bean("fanoutQueue")
12.     public Queue queue1() {
13.         return new Queue("fanout-buget-queue");
14.     }
16.     //注入队列
17.     @Bean("fanoutQueue2")
18.     public Queue queue2() {
19.         return new Queue("fanout-buget-queue2");
20.     }
22.     //注入交换机
23.     @Bean
24.     public FanoutExchange fanoutExchange() {
25.         //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
26.         return new FanoutExchange("fanout-buget-exchange", true, true);
27.     }
29.     //绑定队列和交换机
30.     @Bean("fanoutBinding1")
31.     public Binding binding() {
32.         return BindingBuilder.bind(queue1()).to(fanoutExchange());
33.     }
35.     //绑定队列和交换机
36.     @Bean("fanoutBinding2")
37.     public Binding binding2() {
38.         return BindingBuilder.bind(queue2()).to(fanoutExchange());
39.     }
42. }

复制代码

扇形互换机 Fanout Exchange 消息生产代码如下：

1. //Fanout 模式消息发送
2. public void sendFanoutMessage(String message) {
3.         rabbitTemplate.convertAndSend("fanout-buget-exchange", "", message);
4. }

复制代码

扇形互换机 Fanout Exchange 消息消费代码如下：

1. //Fanout 模式消费端
2. @RabbitListener(queues = "fanout-buget-queue")
3. public void fanoutConsumer1(String message) {
4.         System.out.println("fanout fanout-buget-queue 消息消费成功,message内容为:" + message);
5. }
7. //Fanout 模式消费端
8. @RabbitListener(queues = "fanout-buget-queue2")
9. public void fanoutConsumer2(String message) {
10.         System.out.println("fanout fanout-buget-queue2 消息消费成功,message内容为:" + message);
11. }

复制代码

扇形互换机 Fanout Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-fanout-buget-message")
2. private String sendFanoutBugetMessage(@RequestParam String message) {
3.         myRabbitProducer.sendFanoutMessage(message);
4.         return "OK";
5. }

复制代码

扇形互换机 Fanout Exchange 消息测试（触发消息生产及消费）效果如下：

1. fanout fanout-buget-queue2 消息消费成功,message内容为:hello fanout
2. fanout fanout-buget-queue 消息消费成功,message内容为:hello fanout

复制代码

扇形互换机 Fanout Exchange 队列 Queue 只要与其绑定，就可以把消息路由到对应的 Queue 上并完成消费，效果符合预期。
Headers Exchange（头互换机）
Header Exchange 不依靠 RoutingKey 的判定，而是根据发送的消息内容中的 headers 属性进行匹配，当消息投递到首部互换器时，RabbitMQ 会获取到该消息的 headers，对比其中的键值对是否完全匹配队列和互换器绑定时指定的键值对，假如完全匹配，则消息会路由到该队列，否则不会路由到该队列。
头互换机 Headers Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. import java.util.HashMap;
8. import java.util.Map;
10. @Configuration
11. public class RabbitHeadersConfig {
13.     //注入队列
14.     @Bean("headersQueue")
15.     public Queue queue1() {
16.         return new Queue("headers-buget-queue");
17.     }
19.     //注入队列
20.     @Bean("headersQueue2")
21.     public Queue queue2() {
22.         return new Queue("headers-buget-queue2");
23.     }
25.     //注入交换机
26.     @Bean
27.     public HeadersExchange headersExchange() {
28.         //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
29.         return new HeadersExchange("headers-buget-exchange", true, true);
30.     }
32.     //绑定队列和交换机 绑定 Header中 header-key1 = a 的队列。
33.     @Bean("headersBinding1")
34.     public Binding binding() {
35.         return BindingBuilder.bind(queue1()).to(headersExchange()).where("header-key1").matches("a");
36.     }
38.     //绑定队列和交换机 绑定 Header中  header-key2 = b 的队列。
39.     @Bean("headersBinding2")
40.     public Binding binding2() {
41.         Map<String, Object> map = new HashMap<>();
42.         map.put("header-key2", "b");
43.         return BindingBuilder.bind(queue2()).to(headersExchange()).whereAny(map).match();
44.     }
47. }

复制代码

头互换机 Headers Exchange 消息生产代码如下：

1. //Headers 模式消息发送
2. public void sendHeadersMessage(String message) {
3.         MessageProperties messageProperties = new MessageProperties();
4.         messageProperties.setHeader("header-key1", "a");
5.         rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties));
6.         MessageProperties messageProperties2 = new MessageProperties();
7.         messageProperties2.setHeader("header-key2", "b");
8.         rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties2));
9. }

复制代码

头互换机 Headers Exchange 消息消费代码如下：

1. //headers 模式消费端
2. @RabbitListener(queues = "headers-buget-queue")
3. public void headersConsumer(String message) {
4.         System.out.println("headers headers-buget-queue 消息消费成功,message内容为:" + message);
5. }
7. //headers 模式消费端
8. @RabbitListener(queues = "headers-buget-queue2")
9. public void headersConsumer2(String message) {
10.         System.out.println("headers headers-buget-queue2 消息消费成功,message内容为:" + message);
11. }

复制代码

头互换机 Headers Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-headers-buget-message")
2. private String sendHeadersBugetMessage(@RequestParam String message) {
3.         myRabbitProducer.sendHeadersMessage(message);
4.         return "OK";
5. }

复制代码

头互换机 Headers Exchange 消息测试（触发消息生产及消费）效果如下：

1. headers headers-buget-queue2 消息消费成功,message内容为:hello headers
2. headers headers-buget-queue 消息消费成功,message内容为:hello headers
3. headers headers-buget-queue2 消息消费成功,message内容为:hello headers

复制代码

效果符合预期。
Backup Exchange（备份互换机）
RabbitMQ 本身是不存在备份互换机类型的，备份互换机是抽象出来的一个概念，通过设置互换机的 alternate-exchange 的屬性，属性值是互换机的名称，设置当前互换机的备份互换机，当消息路由无法在当前互换机匹配到合适的队列投递时，将会把消息转到备份互换机，分发到其绑定的备份队列中，备份互换机一般利用扇形互换机，因为其不需要进行路由匹配。
备份互换机 Backup Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. import java.util.HashMap;
8. import java.util.Map;
10. @Configuration
11. public class RabbitBackupConfig {
13.     //注入队列 durable:重启后是否有效 exclusive : 是否独自的 autodelete: 长期未使用是否删除掉
14.     @Bean("backupQueue")
15.     public Queue backupQueue() {
16.         return new Queue("backup-buget-queue", true, false, false);
17.     }
19.     //注入队列
20.     @Bean("noBbackupQueue")
21.     public Queue noBbackupQueue() {
22.         return new Queue("nobackup-buget-queue", true, false, false);
23.     }
25.     //注入 direct 交换机
26.     @Bean("nonBackupExchange")
27.     public DirectExchange nonBackupExchange() {
28.         Map<String, Object> map = new HashMap<>();
29.         //当消息路由无法在当前交换机匹配到合适的队列投递时 将消息转到备份交换机 backup-buget-exchange 分发到其绑定的备份队列中
30.         map.put("alternate-exchange", "backup-buget-exchange");
31.         return new DirectExchange("nobackup-buget-exchange", true, false, map);
32.     }
34.     //注入 Fanout 交换机
35.     @Bean("backupExchange")
36.     public FanoutExchange backupExchange(){
37.         return new FanoutExchange("backup-buget-exchange",true,false);
38.     }
40.     //绑定非备份队列 direct交换机
41.     @Bean
42.     public Binding noBindBackupQueue() {
43.         return BindingBuilder.bind(noBbackupQueue()).to(nonBackupExchange()).with("noback-buget-exchange-routing-key");
44.     }
46.     //扇形交换机的特性决定了它适合做备份交换机 (只要扇形交换机收到消息后就会被转发到与之绑定的队列中 不进行路由判断)
47.     @Bean
48.     public Binding bindBackupQueue(){
49.         return BindingBuilder.bind(backupQueue()).to(backupExchange());
50.     }
53. }

复制代码

备份互换机 Backup Exchange 消息生产代码如下：

1. //备份模式 消息发送
2. public void sendBackupMessage() {
3.         //路由正确匹配 消息投递到非备份队列中
4.         rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key", "hello noBackup");
5.         //路由无法匹配 消息投递到备份队列中 nobackup-buget-exchange 交换机中没有路由 noback-buget-exchange-routing-key-1
6.         rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key-1", "hello backup");
7. }

复制代码

备份互换机 Backup Exchange 消息消费代码如下：

1. //backup 模式消费端(能够正确匹配到的队列)
2. @RabbitListener(queues = "nobackup-buget-queue")
3. public void noBackupConsumer(String message) {
4.         System.out.println("nobackup-buget-queue 消息消费成功,message内容为:" + message);
5. }
7. //backup 模式消费端 (没有匹配到全部来到 backup-buget-queue 队列) 也就是备份队列
8. @RabbitListener(queues = "backup-buget-queue")
9. public void nackupConsumer(String message) {
10.         System.out.println("backup-buget-queue 消息消费成功,message内容为:" + message);
11. }

复制代码

备份互换机 Backup Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-backup-buget-message")
2. private String sendBackupBugetMessage() {
3.         myRabbitProducer.sendBackupMessage();
4.         return "OK";
5. }

复制代码

备份互换机 Backup Exchange 消息测试（触发消息生产及消费）效果如下：

1. nobackup-buget-queue 消息消费成功,message内容为:hello noBackup
2. backup-buget-queue 消息消费成功,message内容为:hello backup

复制代码

我们通过 alternate-exchange 属性给 nobackup-buget-exchange 互换机设置了备份互换机 backup-buget-exchange，我们分别发了两个路由信息 noback-buget-exchange-routing-key 和 noback-buget-exchange-routing-key-1（找不到的路由），最终两条消息被消费了，效果符合预期。
Dead Exchange（死信互换机）
同备份互换机一样 RabbitMQ 本身是不存在死信互换机类型的，死信互换机可以明白成一个拥有特殊意义的直连互换机，通过设置队列中的 x-dead-letter-exchange 和 x-dead-letter-routing-key 属性来设置绑定死信互换机，当消费者拒绝消费、消息积存队列到达最大长度大概消息过期时，消息从正常队列转到死信队列，死信在转移到死信队列时，它的路由是会保存下来，但是假如设置了 x-dead-letter-routing-key 参数的话，路由就会被更换为设置的这个值。
死信互换机 Dead Exchange 和队列 Queue 的路由设置代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.*;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.context.annotation.Configuration;
7. import java.util.HashMap;
8. import java.util.Map;
10. @Configuration
11. public class RabbitDeadConfig {
13.     //注入 死信队列 durable:重启后是否有效 exclusive : 是否独自的 autodelete: 长期未使用是否删除掉
14.     @Bean("deadQueue")
15.     public Queue deadQueue() {
16.         return new Queue("dead-buget-queue", true, false, false);
17.     }
19.     //注入正常队列队列
20.     @Bean("normalQueue")
21.     public Queue normalQueue() {
22.         Map<String, Object> map = new HashMap<>();
23.         //当前队列中 message 过期时间
24.         map.put("x-message-ttl", 5000);
25.         //给当前队列绑定死信交换机
26.         map.put("x-dead-letter-exchange", "dead-buget-exchange");
27.         //绑定 Routing key
28.         map.put("x-dead-letter-routing-key", "dead-buget-exchange-routing-key");
29.         return new Queue("normal-buget-queue", true, false, false, map);
30.     }
32.     //注入 normalExchange 交换机
33.     @Bean("normalExchange")
34.     public DirectExchange normalExchange() {
35.         return new DirectExchange("normal-buget-exchange", true, false);
36.     }
38.     //注入 死信交换机
39.     @Bean("deadExchange")
40.     public DirectExchange deadExchange() {
41.         return new DirectExchange("dead-buget-exchange", true, false);
42.     }
44.     //死信队列和交换器绑定
45.     @Bean
46.     public Binding bindDeadQueue() {
47.         return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead-buget-exchange-routing-key");
48.     }
50.     //正常队列的交换器和队列绑定
51.     @Bean
52.     public Binding bindNormalQueue() {
53.         return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normal-buget-exchange-routing-key");
54.     }
57. }

复制代码

死信互换机 Dead Exchange 消息生产代码如下：

1. //死信 模式消息发送
2. public void sendDeadMessage(String message) {
3.         System.out.println("normal-buget-queue 开始发送,当前时间:"+System.currentTimeMillis());
4.         rabbitTemplate.convertAndSend("normal-buget-exchange", "normal-buget-exchange-routing-key", message);
5. }

复制代码

死信互换机 Dead Exchange 消息消费代码如下：

1. //死信 模式消费端
2. @RabbitListener(queues = "dead-buget-queue")
3. public void deadConsumer(String message) {
4.         System.out.println("dead-buget-queue 开始消费,当前时间:"+System.currentTimeMillis());
5.         System.out.println("dead-buget-queue 消息消费成功,message内容为:" + message);
6. }

复制代码

死信互换机 Dead Exchange 消息测试（触发消息生产及消费）代码代码如下：

1. @GetMapping("/send-dead-buget-message")
2. private String sendBackupBugetMessage(@RequestParam String message) {
3.         myRabbitProducer.sendDeadMessage(message);
4.         return "OK";
5. }

复制代码

死信互换机 Dead Exchange 消息测试（触发消息生产及消费）效果如下：

1. normal-buget-queue 开始发送,当前时间:1725281520323
2. dead-buget-queue 开始消费,当前时间:1725281525327
3. dead-buget-queue 消息消费成功,message内容为:dead

复制代码

我们对 normalQueue 设置了 TTL 为 5秒钟，并为之设置了 死信互换机 dead-buget-exchange 和路由 dead-buget-exchange-routing-key，然后在 normalQueue 上发送了消息，但是没有对 normalQueue 进行监听消费，我们发现过了（1725281525327-1725281520323=5004）5秒后，dead-buget-queue 完成了消息消费，效果符合预期。
消息生产完备代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.core.Message;
4. import org.springframework.amqp.core.MessageProperties;
5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
6. import org.springframework.beans.factory.annotation.Autowired;
7. import org.springframework.stereotype.Component;
9. import java.nio.charset.StandardCharsets;
11. @Component
12. public class MyRabbitProducer {
14.     @Autowired
15.     private RabbitTemplate rabbitTemplate;
17.     //direct 模式消息发送
18.     public void sendDirectMessage(String message) {
19.         rabbitTemplate.convertAndSend("direct-buget-exchange", "direct-buget-exchange-routing-key", message);
20.     }
22.     //topic 模式消息发送
23.     public void sendTopicMessage(String message) {
24.         rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1", message);
25.     }
27.     //topic 模式消息发送
28.     public void sendTopicMessage2(String message) {
29.         rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1.1", message);
30.     }
32.     //Fanout 模式消息发送
33.     public void sendFanoutMessage(String message) {
34.         rabbitTemplate.convertAndSend("fanout-buget-exchange", "", message);
35.     }
37.     //Headers 模式消息发送
38.     public void sendHeadersMessage(String message) {
39.         MessageProperties messageProperties = new MessageProperties();
40.         messageProperties.setHeader("header-key1", "a");
41.         rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties));
42.         MessageProperties messageProperties2 = new MessageProperties();
43.         messageProperties2.setHeader("header-key2", "b");
44.         rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties2));
45.     }
47.     //备份模式 消息发送
48.     public void sendBackupMessage() {
49.         //路由正确匹配 消息投递到非备份队列中
50.         rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key", "hello noBackup");
51.         //路由无法匹配 消息投递到备份队列中 nobackup-buget-exchange 交换机中没有路由 noback-buget-exchange-routing-key-1
52.         rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key-1", "hello backup");
53.     }
55.     //死信 模式消息发送
56.     public void sendDeadMessage(String message) {
57.         System.out.println("normal-buget-queue 开始发送,当前时间:"+System.currentTimeMillis());
58.         rabbitTemplate.convertAndSend("normal-buget-exchange", "normal-buget-exchange-routing-key", message);
59.     }
61. }

复制代码

消息消费完备代码如下：

1. package com.user.service.rabbitmq;
3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
4. import org.springframework.stereotype.Component;
6. @Component
7. public class MyRabbitConsumer {
9.     //direct 直连模式消费端
10.     @RabbitListener(queues = "direct-buget-queue")
11.     public void directConsumer(String message) {
12.         System.out.println("direct 消息消费成功,message内容为:" + message);
13.     }
15.     //topic 模式消费端
16.     @RabbitListener(queues = "topic-buget-queue")
17.     public void topicConsumer(String message) {
18.         System.out.println("topic topic-buget-queue 消息消费成功,message内容为:" + message);
19.     }
21.     //topic 模式消费端
22.     @RabbitListener(queues = "topic-buget-queue2")
23.     public void topicConsumer2(String message) {
24.         System.out.println("topic topic-buget-queue2 消息消费成功,message内容为:" + message);
25.     }
27.     //Fanout 模式消费端
28.     @RabbitListener(queues = "fanout-buget-queue")
29.     public void fanoutConsumer1(String message) {
30.         System.out.println("fanout fanout-buget-queue 消息消费成功,message内容为:" + message);
31.     }
33.     //Fanout 模式消费端
34.     @RabbitListener(queues = "fanout-buget-queue2")
35.     public void fanoutConsumer2(String message) {
36.         System.out.println("fanout fanout-buget-queue2 消息消费成功,message内容为:" + message);
37.     }
39.     //headers 模式消费端
40.     @RabbitListener(queues = "headers-buget-queue")
41.     public void headersConsumer(String message) {
42.         System.out.println("headers headers-buget-queue 消息消费成功,message内容为:" + message);
43.     }
45.     //headers 模式消费端
46.     @RabbitListener(queues = "headers-buget-queue2")
47.     public void headersConsumer2(String message) {
48.         System.out.println("headers headers-buget-queue2 消息消费成功,message内容为:" + message);
49.     }
51.     //backup 模式消费端(能够正确匹配到的队列)
52.     @RabbitListener(queues = "nobackup-buget-queue")
53.     public void noBackupConsumer(String message) {
54.         System.out.println("nobackup-buget-queue 消息消费成功,message内容为:" + message);
55.     }
57.     //backup 模式消费端 (没有匹配到全部来到 backup-buget-queue 队列) 也就是备份队列
58.     @RabbitListener(queues = "backup-buget-queue")
59.     public void nackupConsumer(String message) {
60.         System.out.println("backup-buget-queue 消息消费成功,message内容为:" + message);
61.     }
63.     //死信 模式消费端
64.     @RabbitListener(queues = "dead-buget-queue")
65.     public void deadConsumer(String message) {
66.         System.out.println("dead-buget-queue 开始消费,当前时间:"+System.currentTimeMillis());
67.         System.out.println("dead-buget-queue 消息消费成功,message内容为:" + message);
68.     }
70. }

复制代码

消息测试（触发消息生产及消费）代码代码如下：

1. package com.user.service.controller;
3. import com.user.service.rabbitmq.MyRabbitProducer;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.web.bind.annotation.GetMapping;
6. import org.springframework.web.bind.annotation.RequestParam;
7. import org.springframework.web.bind.annotation.RestController;
9. @RestController
10. public class RabbitController {
12.     @Autowired
13.     private MyRabbitProducer myRabbitProducer;
15.     @GetMapping("/send-direct-buget-message")
16.     private String sendDirectBugetMessage(@RequestParam String message) {
17.         myRabbitProducer.sendDirectMessage(message);
18.         return "OK";
19.     }
21.     @GetMapping("/send-topic-buget-message")
22.     private String sendTopicBugetMessage(@RequestParam String message) {
23.         myRabbitProducer.sendTopicMessage(message);
24.         myRabbitProducer.sendTopicMessage2(message);
25.         return "OK";
26.     }
28.     @GetMapping("/send-fanout-buget-message")
29.     private String sendFanoutBugetMessage(@RequestParam String message) {
30.         myRabbitProducer.sendFanoutMessage(message);
31.         return "OK";
32.     }
34.     @GetMapping("/send-headers-buget-message")
35.     private String sendHeadersBugetMessage(@RequestParam String message) {
36.         myRabbitProducer.sendHeadersMessage(message);
37.         return "OK";
38.     }
40.     @GetMapping("/send-backup-buget-message")
41.     private String sendBackupBugetMessage() {
42.         myRabbitProducer.sendBackupMessage();
43.         return "OK";
44.     }
46.     @GetMapping("/send-dead-buget-message")
47.     private String sendBackupBugetMessage(@RequestParam String message) {
48.         myRabbitProducer.sendDeadMessage(message);
49.         return "OK";
50.     }
53. }

复制代码

尚硅谷 RabbitMQ 学习视频
总结：本篇简单分享了 RabbitMQ 六种互换机的消息生产和消费的利用，只是 Demo 案例，在实际项目中更换成业务代码即可，盼望可以资助到有需要的朋侪。
如有不准确的地方请各位指出纠正。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。