论坛 › 数据仓库与分析 › Spring Boot 整合 RabbitMQ 详解

民工心事 发表于 2024-11-29 04:26:56

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 设置如下：
spring.rabbitmq.host= xxx.xxx.xxx
spring.rabbitmq.port= 5672
spring.rabbitmq.username= admin
spring.rabbitmq.password= admin
spring.rabbitmq.virtual-host = /study
项目 pom.xml 文件中引入 spring-boot-starter-amqp 依靠如下：
<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.4.5</version>
</dependency>
RabbitMQ 利用

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

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitDirectConfig {

    //注入队列
    @Bean
    public Queue queue() {
      return new Queue("direct-buget-queue");
    }

    //注入交换机
    @Bean
    public DirectExchange directExchange() {
      //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
      return new DirectExchange("direct-buget-exchange", true, true);
    }

    //绑定队列和交换机
    @Bean
    public Binding binding() {
      return BindingBuilder.bind(queue()).to(directExchange()).with("direct-buget-exchange-routing-key");
    }

}
直连互换机 Direct Exchange 消息生产代码如下：
//direct 模式消息发送
public void sendDirectMessage(String message) {
        rabbitTemplate.convertAndSend("direct-buget-exchange", "direct-buget-exchange-routing-key", message);
}
直连互换机 Direct Exchange 消息消费代码如下：
//direct 直连模式消费端
@RabbitListener(queues = "direct-buget-queue")
public void directConsumer(String message) {
        System.out.println("direct 消息消费成功,message内容为:" + message);
}

直连互换机 Direct Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-direct-buget-message")
private String sendDirectBugetMessage(@RequestParam String message) {
        myRabbitProducer.sendDirectMessage(message);
        return "OK";
}
直连互换机 Direct Exchange 消息测试（触发消息生产及消费）效果如下：
direct 消息消费成功,message内容为:hello message
直连互换机，一对一模式，效果符合预期。
Topic Exchange（主题互换机）
主题互换机支持路由模糊匹配，可以利用星号和井号（#）作为通配符进行匹配，其中 ”*“ 可以代替一个单词，（#） 可以代替恣意个单词。
主题互换机 Topic Exchange 和队列 Queue 的路由设置代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitTopicConfig {

    //注入队列
    @Bean("topicQueue")
    public Queue queue1() {
      return new Queue("topic-buget-queue");
    }

    //注入队列
    @Bean("topicQueue2")
    public Queue queue2() {
      return new Queue("topic-buget-queue2");
    }

    //注入交换机
    @Bean
    public TopicExchange topicExchange() {
      //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
      return new TopicExchange("topic-buget-exchange", true, true);
    }

    //绑定队列和交换机
    @Bean("topicBinding")
    public Binding binding() {
      return BindingBuilder.bind(queue1()).to(topicExchange()).with("topic.buget.exchange.routing.key.*");
    }
    //绑定队列和交换机
    @Bean("topicBinding2")
    public Binding binding2() {
      return BindingBuilder.bind(queue2()).to(topicExchange()).with("topic.buget.exchange.routing.key.#");
    }

}
主题互换机 Topic Exchange 消息生产代码如下：
//topic 模式消息发送
public void sendTopicMessage(String message) {
        rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1", message);
}

//topic 模式消息发送
public void sendTopicMessage2(String message) {
        rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1.1", message);
}
主题互换机 Topic Exchange 消息消费代码如下：
//topic 模式消费端
@RabbitListener(queues = "topic-buget-queue")
public void topicConsumer(String message) {
        System.out.println("topic topic-buget-queue 消息消费成功,message内容为:" + message);
}

//topic 模式消费端
@RabbitListener(queues = "topic-buget-queue2")
public void topicConsumer2(String message) {
        System.out.println("topic topic-buget-queue2 消息消费成功,message内容为:" + message);
}

主题互换机 Topic Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-topic-buget-message")
private String sendTopicBugetMessage(@RequestParam String message) {
        myRabbitProducer.sendTopicMessage(message);
        myRabbitProducer.sendTopicMessage2(message);
        return "OK";
}
主题互换机 Topic Exchange 消息测试（触发消息生产及消费）效果如下：
topic topic-buget-queue 消息消费成功,message内容为:hello topic
topic topic-buget-queue2 消息消费成功,message内容为:hello topic
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 的路由设置代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitFanoutConfig {

    //注入队列
    @Bean("fanoutQueue")
    public Queue queue1() {
      return new Queue("fanout-buget-queue");
    }

    //注入队列
    @Bean("fanoutQueue2")
    public Queue queue2() {
      return new Queue("fanout-buget-queue2");
    }

    //注入交换机
    @Bean
    public FanoutExchange fanoutExchange() {
      //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
      return new FanoutExchange("fanout-buget-exchange", true, true);
    }

    //绑定队列和交换机
    @Bean("fanoutBinding1")
    public Binding binding() {
      return BindingBuilder.bind(queue1()).to(fanoutExchange());
    }

    //绑定队列和交换机
    @Bean("fanoutBinding2")
    public Binding binding2() {
      return BindingBuilder.bind(queue2()).to(fanoutExchange());
    }


}
扇形互换机 Fanout Exchange 消息生产代码如下：
//Fanout 模式消息发送
public void sendFanoutMessage(String message) {
        rabbitTemplate.convertAndSend("fanout-buget-exchange", "", message);
}
扇形互换机 Fanout Exchange 消息消费代码如下：
//Fanout 模式消费端
@RabbitListener(queues = "fanout-buget-queue")
public void fanoutConsumer1(String message) {
        System.out.println("fanout fanout-buget-queue 消息消费成功,message内容为:" + message);
}

//Fanout 模式消费端
@RabbitListener(queues = "fanout-buget-queue2")
public void fanoutConsumer2(String message) {
        System.out.println("fanout fanout-buget-queue2 消息消费成功,message内容为:" + message);
}

扇形互换机 Fanout Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-fanout-buget-message")
private String sendFanoutBugetMessage(@RequestParam String message) {
        myRabbitProducer.sendFanoutMessage(message);
        return "OK";
}
扇形互换机 Fanout Exchange 消息测试（触发消息生产及消费）效果如下：
fanout fanout-buget-queue2 消息消费成功,message内容为:hello fanout
fanout fanout-buget-queue 消息消费成功,message内容为:hello fanout
扇形互换机 Fanout Exchange 队列 Queue 只要与其绑定，就可以把消息路由到对应的 Queue 上并完成消费，效果符合预期。
Headers Exchange（头互换机）
Header Exchange 不依靠 RoutingKey 的判定，而是根据发送的消息内容中的 headers 属性进行匹配，当消息投递到首部互换器时，RabbitMQ 会获取到该消息的 headers，对比其中的键值对是否完全匹配队列和互换器绑定时指定的键值对，假如完全匹配，则消息会路由到该队列，否则不会路由到该队列。
头互换机 Headers Exchange 和队列 Queue 的路由设置代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitHeadersConfig {

    //注入队列
    @Bean("headersQueue")
    public Queue queue1() {
      return new Queue("headers-buget-queue");
    }

    //注入队列
    @Bean("headersQueue2")
    public Queue queue2() {
      return new Queue("headers-buget-queue2");
    }

    //注入交换机
    @Bean
    public HeadersExchange headersExchange() {
      //durable:重启后是否有效 autodelete: 长期未使用是否删除掉
      return new HeadersExchange("headers-buget-exchange", true, true);
    }

    //绑定队列和交换机 绑定 Header中 header-key1 = a 的队列。
    @Bean("headersBinding1")
    public Binding binding() {
      return BindingBuilder.bind(queue1()).to(headersExchange()).where("header-key1").matches("a");
    }

    //绑定队列和交换机 绑定 Header中header-key2 = b 的队列。
    @Bean("headersBinding2")
    public Binding binding2() {
      Map<String, Object> map = new HashMap<>();
      map.put("header-key2", "b");
      return BindingBuilder.bind(queue2()).to(headersExchange()).whereAny(map).match();
    }


}
头互换机 Headers Exchange 消息生产代码如下：
//Headers 模式消息发送
public void sendHeadersMessage(String message) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("header-key1", "a");
        rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties));
        MessageProperties messageProperties2 = new MessageProperties();
        messageProperties2.setHeader("header-key2", "b");
        rabbitTemplate.convertAndSend("headers-buget-exchange", "", new Message(message.getBytes(StandardCharsets.UTF_8), messageProperties2));
}

头互换机 Headers Exchange 消息消费代码如下：
//headers 模式消费端
@RabbitListener(queues = "headers-buget-queue")
public void headersConsumer(String message) {
        System.out.println("headers headers-buget-queue 消息消费成功,message内容为:" + message);
}

//headers 模式消费端
@RabbitListener(queues = "headers-buget-queue2")
public void headersConsumer2(String message) {
        System.out.println("headers headers-buget-queue2 消息消费成功,message内容为:" + message);
}
头互换机 Headers Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-headers-buget-message")
private String sendHeadersBugetMessage(@RequestParam String message) {
        myRabbitProducer.sendHeadersMessage(message);
        return "OK";
}

头互换机 Headers Exchange 消息测试（触发消息生产及消费）效果如下：
headers headers-buget-queue2 消息消费成功,message内容为:hello headers
headers headers-buget-queue 消息消费成功,message内容为:hello headers
headers headers-buget-queue2 消息消费成功,message内容为:hello headers
效果符合预期。
Backup Exchange（备份互换机）
RabbitMQ 本身是不存在备份互换机类型的，备份互换机是抽象出来的一个概念，通过设置互换机的 alternate-exchange 的屬性，属性值是互换机的名称，设置当前互换机的备份互换机，当消息路由无法在当前互换机匹配到合适的队列投递时，将会把消息转到备份互换机，分发到其绑定的备份队列中，备份互换机一般利用扇形互换机，因为其不需要进行路由匹配。
备份互换机 Backup Exchange 和队列 Queue 的路由设置代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitBackupConfig {

    //注入队列 durable:重启后是否有效 exclusive : 是否独自的 autodelete: 长期未使用是否删除掉
    @Bean("backupQueue")
    public Queue backupQueue() {
      return new Queue("backup-buget-queue", true, false, false);
    }

    //注入队列
    @Bean("noBbackupQueue")
    public Queue noBbackupQueue() {
      return new Queue("nobackup-buget-queue", true, false, false);
    }

    //注入 direct 交换机
    @Bean("nonBackupExchange")
    public DirectExchange nonBackupExchange() {
      Map<String, Object> map = new HashMap<>();
      //当消息路由无法在当前交换机匹配到合适的队列投递时 将消息转到备份交换机 backup-buget-exchange 分发到其绑定的备份队列中
      map.put("alternate-exchange", "backup-buget-exchange");
      return new DirectExchange("nobackup-buget-exchange", true, false, map);
    }

    //注入 Fanout 交换机
    @Bean("backupExchange")
    public FanoutExchange backupExchange(){
      return new FanoutExchange("backup-buget-exchange",true,false);
    }

    //绑定非备份队列 direct交换机
    @Bean
    public Binding noBindBackupQueue() {
      return BindingBuilder.bind(noBbackupQueue()).to(nonBackupExchange()).with("noback-buget-exchange-routing-key");
    }

    //扇形交换机的特性决定了它适合做备份交换机 (只要扇形交换机收到消息后就会被转发到与之绑定的队列中 不进行路由判断)
    @Bean
    public Binding bindBackupQueue(){
      return BindingBuilder.bind(backupQueue()).to(backupExchange());
    }


}
备份互换机 Backup Exchange 消息生产代码如下：
//备份模式 消息发送
public void sendBackupMessage() {
        //路由正确匹配 消息投递到非备份队列中
        rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key", "hello noBackup");
        //路由无法匹配 消息投递到备份队列中 nobackup-buget-exchange 交换机中没有路由 noback-buget-exchange-routing-key-1
        rabbitTemplate.convertAndSend("nobackup-buget-exchange", "noback-buget-exchange-routing-key-1", "hello backup");
}

备份互换机 Backup Exchange 消息消费代码如下：
//backup 模式消费端(能够正确匹配到的队列)
@RabbitListener(queues = "nobackup-buget-queue")
public void noBackupConsumer(String message) {
        System.out.println("nobackup-buget-queue 消息消费成功,message内容为:" + message);
}

//backup 模式消费端 (没有匹配到全部来到 backup-buget-queue 队列) 也就是备份队列
@RabbitListener(queues = "backup-buget-queue")
public void nackupConsumer(String message) {
        System.out.println("backup-buget-queue 消息消费成功,message内容为:" + message);
}
备份互换机 Backup Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-backup-buget-message")
private String sendBackupBugetMessage() {
        myRabbitProducer.sendBackupMessage();
        return "OK";
}
备份互换机 Backup Exchange 消息测试（触发消息生产及消费）效果如下：
nobackup-buget-queue 消息消费成功,message内容为:hello noBackup
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 的路由设置代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitDeadConfig {

    //注入 死信队列 durable:重启后是否有效 exclusive : 是否独自的 autodelete: 长期未使用是否删除掉
    @Bean("deadQueue")
    public Queue deadQueue() {
      return new Queue("dead-buget-queue", true, false, false);
    }

    //注入正常队列队列
    @Bean("normalQueue")
    public Queue normalQueue() {
      Map<String, Object> map = new HashMap<>();
      //当前队列中 message 过期时间
      map.put("x-message-ttl", 5000);
      //给当前队列绑定死信交换机
      map.put("x-dead-letter-exchange", "dead-buget-exchange");
      //绑定 Routing key
      map.put("x-dead-letter-routing-key", "dead-buget-exchange-routing-key");
      return new Queue("normal-buget-queue", true, false, false, map);
    }

    //注入 normalExchange 交换机
    @Bean("normalExchange")
    public DirectExchange normalExchange() {
      return new DirectExchange("normal-buget-exchange", true, false);
    }

    //注入 死信交换机
    @Bean("deadExchange")
    public DirectExchange deadExchange() {
      return new DirectExchange("dead-buget-exchange", true, false);
    }

    //死信队列和交换器绑定
    @Bean
    public Binding bindDeadQueue() {
      return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead-buget-exchange-routing-key");
    }

    //正常队列的交换器和队列绑定
    @Bean
    public Binding bindNormalQueue() {
      return BindingBuilder.bind(normalQueue()).to(normalExchange()).with("normal-buget-exchange-routing-key");
    }


}
死信互换机 Dead Exchange 消息生产代码如下：
//死信 模式消息发送
public void sendDeadMessage(String message) {
        System.out.println("normal-buget-queue 开始发送,当前时间:"+System.currentTimeMillis());
        rabbitTemplate.convertAndSend("normal-buget-exchange", "normal-buget-exchange-routing-key", message);
}
死信互换机 Dead Exchange 消息消费代码如下：
//死信 模式消费端
@RabbitListener(queues = "dead-buget-queue")
public void deadConsumer(String message) {
        System.out.println("dead-buget-queue 开始消费,当前时间:"+System.currentTimeMillis());
        System.out.println("dead-buget-queue 消息消费成功,message内容为:" + message);
}
死信互换机 Dead Exchange 消息测试（触发消息生产及消费）代码代码如下：
@GetMapping("/send-dead-buget-message")
private String sendBackupBugetMessage(@RequestParam String message) {
        myRabbitProducer.sendDeadMessage(message);
        return "OK";
}
死信互换机 Dead Exchange 消息测试（触发消息生产及消费）效果如下：
normal-buget-queue 开始发送,当前时间:1725281520323
dead-buget-queue 开始消费,当前时间:1725281525327
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 完成了消息消费，效果符合预期。
消息生产完备代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Component
public class MyRabbitProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

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

    //topic 模式消息发送
    public void sendTopicMessage(String message) {
      rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1", message);
    }

    //topic 模式消息发送
    public void sendTopicMessage2(String message) {
      rabbitTemplate.convertAndSend("topic-buget-exchange", "topic.buget.exchange.routing.key.1.1", message);
    }

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

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

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

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

}
消息消费完备代码如下：
package com.user.service.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class MyRabbitConsumer {

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

    //topic 模式消费端
    @RabbitListener(queues = "topic-buget-queue")
    public void topicConsumer(String message) {
      System.out.println("topic topic-buget-queue 消息消费成功,message内容为:" + message);
    }

    //topic 模式消费端
    @RabbitListener(queues = "topic-buget-queue2")
    public void topicConsumer2(String message) {
      System.out.println("topic topic-buget-queue2 消息消费成功,message内容为:" + message);
    }

    //Fanout 模式消费端
    @RabbitListener(queues = "fanout-buget-queue")
    public void fanoutConsumer1(String message) {
      System.out.println("fanout fanout-buget-queue 消息消费成功,message内容为:" + message);
    }

    //Fanout 模式消费端
    @RabbitListener(queues = "fanout-buget-queue2")
    public void fanoutConsumer2(String message) {
      System.out.println("fanout fanout-buget-queue2 消息消费成功,message内容为:" + message);
    }

    //headers 模式消费端
    @RabbitListener(queues = "headers-buget-queue")
    public void headersConsumer(String message) {
      System.out.println("headers headers-buget-queue 消息消费成功,message内容为:" + message);
    }

    //headers 模式消费端
    @RabbitListener(queues = "headers-buget-queue2")
    public void headersConsumer2(String message) {
      System.out.println("headers headers-buget-queue2 消息消费成功,message内容为:" + message);
    }

    //backup 模式消费端(能够正确匹配到的队列)
    @RabbitListener(queues = "nobackup-buget-queue")
    public void noBackupConsumer(String message) {
      System.out.println("nobackup-buget-queue 消息消费成功,message内容为:" + message);
    }

    //backup 模式消费端 (没有匹配到全部来到 backup-buget-queue 队列) 也就是备份队列
    @RabbitListener(queues = "backup-buget-queue")
    public void nackupConsumer(String message) {
      System.out.println("backup-buget-queue 消息消费成功,message内容为:" + message);
    }

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

}
消息测试（触发消息生产及消费）代码代码如下：
package com.user.service.controller;

import com.user.service.rabbitmq.MyRabbitProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class RabbitController {

    @Autowired
    private MyRabbitProducer myRabbitProducer;

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

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

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

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

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

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


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

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

查看完整版本: Spring Boot 整合 RabbitMQ 详解