Spring Boot 整合 RabbitMQ 详解

标签：String exchange Spring buget RabbitMQ 交换机 Boot message public

前言：

在消息中间件领域中 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 案例，在实际项目中替换成业务代码即可，希望可以帮助到有需要的朋友。

如有不正确的地方请各位指出纠正。

标签：String,exchange,Spring,buget,RabbitMQ,交换机,Boot,message,public
From： https://blog.csdn.net/weixin_42118323/article/details/141829003

Spring Boot 整合 RabbitMQ 详解

RabbitMQ 使用

相关文章

赞助商

阅读排行