【Spring AMQP】基于 Spring AMQP 实现简单、工作队列模型，发布订阅模型 Fanout、Direct、Topic ，以及更改 AMQP 的消息转换器

小咪咪 2023-10-16 18:00 31阅读 0赞

#### 文章目录 ####

*  前言
 *  一、初识 Spring AMQP
 *   *  1.1 什么是 Spring AMQP
     *  1.2 在 Spring Boot 项目中集成 Spring AMQP
     *  1.3 RabbitTemplate 介绍
 *  二、实现简单队列模型（Basic Queue）
 *   *  2.1 简单队列模型
     *  2.2 消息发布者
     *  2.3 消息消费者
 *  三、实现工作队列模型（Work Queue）
 *   *  3.1 工作队列模型
     *  3.2 消息发布者
     *  3.3 消息消费者
     *  3.4 修改消费预取限制
 *  四、发布（Publish）和订阅（Subscribe）
 *  五、实现 Fanout（广播）发布订阅模型
 *   *  5.1 发布订阅 - Fanout Exchange
     *  5.2 消息发布者
     *  5.3 消息订阅者
 *  六、实现 Direct（路由）发布订阅模型
 *   *  6.1 发布订阅 - Direct Exchange
     *  6.2 消息发布者
     *  6.3 消息订阅者
 *  七、实现 Topic（广播）发布订阅模型
 *   *  7.1 发布订阅 - Topic Exchange
     *  7.2 消息发布者
     *  7.3 消息订阅者
 *  八、AMQP 的消息转换器
 *   *  8.1 测试发送 Object 类型消息
     *  8.2 消息转换器
     *  8.3 更改消息转换器类型

--------------------

## 前言 ##

在分布式系统中，消息队列是一种常见的通信方式，用于实现不同模块之间的解耦、异步通信。Spring AMQP 是 Spring 框架对 AMQP（高级消息队列协议）的支持，提供了丰富的 API 和注解，使得在 Spring 项目中使用消息队列变得更加简便。

本文将介绍如何基于 Spring AMQP 实现简单队列模型、工作队列模型，以及发布订阅模型的 Fanout、Direct、Topic 的实现。同时，我们还会探讨 AMQP 的消息转换器，以及如何更改默认的消息转换器。

## 一、初识 Spring AMQP ##

### 1.1 什么是 Spring AMQP ###

Spring AMQP 是 Spring 框架对 AMQP（高级消息队列协议）的支持。它提供了一套用于发送和接收消息的高级 API，使得在 Spring 项目中使用消息队列变得更加方便。

Spring AMQP 的官方地址：[https://spring.io/projects/spring-amqp][https_spring.io_projects_spring-amqp]。

AMQP 是一种网络协议，它定义了消息的格式和传递方式，允许不同的应用程序通过消息进行通信。Spring AMQP 提供了一个高级的抽象层，隐藏了底层协议的复杂性，使得开发者能够更专注于业务逻辑。

**Spring AMQP 和 AMQP：**  
![b1465671d4ce4a329bd53df9fb872cd6.png][]

### 1.2 在 Spring Boot 项目中集成 Spring AMQP ###

在 Spring Boot 项目中，集成 Spring AMQP 非常简单。只需在 `pom.xml` 文件中添加相关依赖，Spring Boot 会自动配置好所需的 Bean。以下是一个简单的 Spring Boot 项目的 `pom.xml` 文件：

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

这样，我们就引入了 Spring AMQP 相关的依赖，并可以在项目中开始使用 Spring AMQP。

### 1.3 RabbitTemplate 介绍 ###

`RabbitTemplate` 是 Spring AMQP 提供的一个核心类，用于在 Spring 应用中简化 RabbitMQ 操作。它提供了一种方便的方式来发送和接收消息，隐藏了与 RabbitMQ 的低级交互细节，使得开发者能够更轻松地集成消息队列到 Spring 应用中。

以下是 `RabbitTemplate` 的一些主要特性和用法：

1.  **消息发送：** `RabbitTemplate` 提供了 `convertAndSend` 方法，用于将消息发送到指定的队列或交换机。它支持将消息体进行自动序列化，可以发送任何 Java 对象。
    
        rabbitTemplate.convertAndSend("exchange", "routingKey", message);
2.  **消息接收：** 通过使用 `RabbitTemplate` 的 `receive` 或 `receiveAndConvert` 方法，可以从指定的队列接收消息。
    
        Object message = rabbitTemplate.receiveAndConvert("queue");
3.  **消息回调：** `RabbitTemplate` 提供了异步消息发送的方式，可以通过 `sendAndReceive` 方法发送消息并提供一个回调函数，用于处理接收到的响应消息。
    
        rabbitTemplate.sendAndReceive("exchange", "routingKey", message, (Message receivedMessage) -> {
              
            // 处理接收到的响应消息
        });
4.  **消息转换器：** `RabbitTemplate` 使用消息转换器（`MessageConverter`）来实现消息的序列化和反序列化。默认情况下，它使用 `SimpleMessageConverter`。
    
        // 设置自定义消息转换器
        rabbitTemplate.setMessageConverter(myCustomMessageConverter);
5.  **消息确认：** `RabbitTemplate` 支持消息的确认机制，可以通过设置 `CorrelationData` 来实现消息的确认和失败处理。
    
        rabbitTemplate.convertAndSend("exchange", "routingKey", message, new CorrelationData("correlationId"));
6.  **发送确认和回调：** 可以设置 `ConfirmCallback` 和 `ReturnsCallback` 接口来处理发送消息的确认和返回结果。
    
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
              
            // 处理发送确认回调
        });
        
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
              
            // 处理发送失败的消息回调
        });
7.  **其他操作：** `RabbitTemplate` 还提供了其他一些方便的方法，如发送批量消息、获取 RabbitMQ 的连接等。
    
        // 发送批量消息
        rabbitTemplate.convertAndSend("exchange", "routingKey", Arrays.asList(message1, message2, message3));

总体而言，`RabbitTemplate` 是在 Spring 中使用 RabbitMQ 时非常实用的工具，它简化了消息的发送和接收操作，提供了一种更高层次的抽象，使得与 RabbitMQ 的集成更加方便。

## 二、实现简单队列模型（Basic Queue） ##

### 2.1 简单队列模型 ###

在简单队列模型中，有一个生产者将消息发送到一个队列（Queue），然后有一个消费者从队列中接收并处理消息。这是一种基本的消息传递模型，适用于点对点的通信场景。  
![8cab6f5bcf084d388cd00d33f2b1592f.png][]

下面利用 SpringAMQP 实现发送和消费消息的的基础消息队列功能。

因为`publisher`和`consumer`两个模块都需要 AMQP 依赖，因此这里把依赖直接放到父工程 `mq-demo` 的 `pomxml` 中：

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

### 2.2 消息发布者 ###

在`publisher`中编写测试方法，向`simple.queue`发送消息：

1.  在`publisher`服务中编写`application.yml`配置文件，添加 RabbitMQ 连接信息：

# RabbitMQ 相关配置
    spring:
      rabbitmq:
        host: 192.168.146.129 # rabbitMQ 的 IP 地址
        port: 5672 # 端口
        username: admin
        password: 123456
        virtual-host: /

配置说明：

*  `host`: RabbitMQ 服务器的 IP 地址。
 *  `port`: RabbitMQ 服务器的端口。
 *  `username` 和 `password`: 连接 RabbitMQ 时的用户名和密码。
 *  `virtual-host`: RabbitMQ 的虚拟主机。

1.  在`publisher`服务中新建一个测试类，编写测试方法：

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class SpringAmqpTest {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Test
        public void testSendMessageToSimpleQueue() {
        
            String queueName = "simple.queue";
            String message = "Hello, Spring AMQP!";
            rabbitTemplate.convertAndSend(queueName, message);
        }
    }

这段代码是一个测试类，用于验证在 Spring Boot 应用中通过 RabbitTemplate 发送消息到 RabbitMQ 的简单队列（`simple.queue`）中。

**解释每个部分的作用：**

*  `@RunWith(SpringRunner.class)`: 这是 JUnit 的注解，它告诉 JUnit 使用 Spring 的测试支持。
 *  `@SpringBootTest`: 这是 Spring Boot 的注解，用于指示 JUnit 运行时加载 Spring Boot 的应用上下文。
 *  `@Autowired private RabbitTemplate rabbitTemplate;`: 这是使用 Spring 的依赖注入，将 RabbitTemplate 注入到测试类中。RabbitTemplate 是 Spring AMQP 提供的用于发送和接收消息的核心类。
 *  `@Test public void testSendMessageToSimpleQueue() { ... }`: 这是一个测试方法。在这个方法中，我们创建了一个队列名称 `simple.queue`，然后使用 RabbitTemplate 的 `convertAndSend` 方法发送一条消息。
    
     *  `queueName`: 定义了要发送消息的目标队列的名称，即 `simple.queue`。
     *  `message`: 要发送的消息内容，即 “Hello, Spring AMQP!”。

这个测试方法的目的是确保我们能够使用 RabbitTemplate 成功地将消息发送到 RabbitMQ 中的 `simple.queue` 队列中。这是一个基本的测试用例，用于验证消息的生产者功能。

### 2.3 消息消费者 ###

在 `consumer` 服务中，我们需要编写代码来监听 `simple.queue` 队列并处理收到的消息。

1.  首先编写 `application.yml` 配置文件，添加 RabbitMQ 连接信息：

# RabbitMQ 相关配置
    spring:
      rabbitmq:
        host: 192.168.146.129 # RabbitMQ 的 IP 地址
        port: 5672 # 端口
        username: lhf
        password: 123456
        virtual-host: /

1.  现在，我们来编写一个简单的消息处理类 `SpringRabbitMQListener`，然后在里面编写一个方法用于监听 `simple.queue` 队列：

@Component
    public class SpringRabbitMQListener {
        
    
        @RabbitListener(queues = "simple.queue")
        public void listenSimpleQueue(String message) {
        
            System.out.println("消费者收到 simple.queue 的消息：【" + message + "】");
        }
    }

这个类使用 `@RabbitListener` 注解标记了一个方法 `listenSimpleQueue`，该方法会在接收到来自 `simple.queue` 队列的消息时被调用。在这个例子中，我们简单地将消息打印到控制台。

以上是一个简单队列模型的实现示例。当 `publisher` 服务发送消息到 `simple.queue` 队列时，`consumer` 服务中的 `SpringRabbitMQListener` 类会监听到消息并进行处理。可以通过查看控制台日志来验证消息是否被正确接收。

![6636463a4140467f985adecdaf3f6139.png][]

## 三、实现工作队列模型（Work Queue） ##

### 3.1 工作队列模型 ###

工作队列模型（Work Queue Model）是一种消息处理机制，通常用于提高系统的消息处理速度，防止消息堆积。该模型基于队列数据结构，通过将任务（或消息）按顺序排列在队列中，使得系统能够有序地处理它们。

**模型概述：**

*  **生产者（Producer）：** 生产者应用产生消息并将其发送到工作队列。
 *  **工作队列（Work Queue）：** 工作队列是消息的缓冲区，用于存储生产者发送的消息。多个消费者可以共享同一个工作队列。
 *  **消费者（Consumer）：** 消费者应用从工作队列中接收并处理消息。多个消费者可以同时从工作队列中接收消息，实现任务的并发处理。

**模型图解：**

![86b36d59efb14f268c8f5acc437a9d9b.png][]

**工作流程：**

1.  **生产者发送消息：** 生产者应用产生消息，并将这些消息发送到工作队列。在图中，“publisher” 发送消息到 “queue”。
2.  **工作队列存储消息：** 工作队列接收到生产者发送的消息，并将这些消息存储在队列中。消息在队列中等待被消费。
3.  **多个消费者接收并处理消息：** 多个消费者（“consumer1” 和 “consumer2”）同时从工作队列中接收消息，并处理这些消息。每个消息只会被一个消费者处理，从而实现负载均衡。

**特点和优势：**

*  **消息处理速度提高：** 多个消费者可以并行处理消息，从而提高整体的消息处理速度。
 *  **避免消息堆积：** 即使有大量消息进入工作队列，多个消费者可以并发地处理这些消息，防止消息在队列中堆积。

工作队列模型的灵活性和可扩展性使其成为处理大量任务和提高系统处理能力的有效手段。

### 3.2 消息发布者 ###

现在使用一个测试方法来模拟消息发布者，再这个方法中，每秒发送 50 条消息到 `simplie.queue` 队列中：

// 生产者：每秒生成 50 条消息
        @Test
        public void testSendMessageToWorkQueue() throws InterruptedException {
        
            String queueName = "simple.queue";
            String message = "Hello, message__!";
    
            for (int i = 1; i <= 50; i++) {
        
                rabbitTemplate.convertAndSend(queueName, message + i);
                Thread.sleep(20);
            }
    
        }

**对上述代码的说明：**

*  `@Test` 注解表示这是一个测试方法。
 *  `rabbitTemplate.convertAndSend(queueName, message + i)` 使用 RabbitTemplate 发送消息到名为 “simple.queue” 的队列。
 *  在循环中，生产者每秒生成 50 条消息，并通过 `Thread.sleep(20)` 模拟消息的产生速度。

这段代码的目的是模拟消息的生成过程，为后续的消费者处理提供消息源。

### 3.3 消息消费者 ###

在 `SpringRabbitMQListener` 消息处理类中，创建两个方法来模拟 `consumer1` 和 `consumer2`，其中`consumer1` 每秒能够处理 50 条消息，而 `consumer2`每秒能够处理 10 条消息：

// 消费者1：每秒处理 50 条消息
        @RabbitListener(queues = "simple.queue")
        public void listenWorkQueue1(String message) throws InterruptedException {
        
            System.out.println("消费者1 收到 simple.queue 的消息：【" + message + "】" + LocalTime.now());
            Thread.sleep(20);
        }
    
        // 消费者2：每秒处理 10 条消息
        @RabbitListener(queues = "simple.queue")
        public void listenWorkQueue2(String message) throws InterruptedException {
        
            System.err.println("消费者2 收到 simple.queue 的消息：【" + message + "】" + LocalTime.now());
            Thread.sleep(100);
        }

**对上述代码的说明：**

*  `@RabbitListener(queues = "simple.queue")` 注解指定了这两个方法监听的队列为 “simple.queue”。
 *  `listenWorkQueue1` 方法模拟消费者1，通过 `System.out.println` 输出消息接收信息，并通过 `Thread.sleep(20)` 模拟每秒处理 50 条消息的速度。
 *  `listenWorkQueue2` 方法模拟消费者2，通过 `System.err.println` 输出消息接收信息（红色），并通过 `Thread.sleep(100)` 模拟每秒处理 10 条消息的速度。

这两个消费者方法共同实现了对 simple.queue 中消息的并发处理，其中消费者1每秒处理 50 条消息，消费者2每秒处理 10 条消息。

### 3.4 修改消费预取限制 ###

当完成上述工作队列模型的生产和消费代码之后，我们就可以尝试运行一下。理论上来说，`consumer1` 处理的消息数量是 `consumer2` 的 5 倍，看看是否达到预期的效果：

![在这里插入图片描述][8890cbd767df4f92a45ef564e3502df9.png]  
此时发现，实际上消费者1处理的是编号为奇数的消息，而消费者2处理的则是奇数为偶数的信息，二者分配的消息数目相等，原因在于Spring AMQP 采用的是消费预存，即二者都是先平均分配所有的消息到自己的缓冲区中，然后再各自处理。

在工作队列模型中，我们希望消费者能够按照一定的策略来获取消息，以实现负载均衡。默认情况下，Spring AMQP 使用了预取机制（prefetch），即消费者从队列中预先获取一定数量的消息，然后再处理。这个数量可以通过配置进行修改。

在上述工作队列模型的实现中，我们可以通过修改 `application.yml` 文件来调整消息预取的设置：

spring:
      rabbitmq:
        host: 192.168.146.129 # rabbitMQ 的 IP 地址
        port: 5672 # 端口
        username: lisi
        password: 123456
        virtual-host: /
        listener:
          simple:
            prefetch: 1 # 修改消息预取，每次只获取一条信息，处理完成之后才能获取下一条消息

上述配置中，将 `spring.rabbitmq.listener.simple.prefetch` 设置为 1，表示每个消费者每次预取一条消息。这样，消费者在处理完当前消息之后，才能预取并处理下一条消息。这种方式可以确保消息在消费者之间更均匀地分配，实现了一定程度的负载均衡。

**再次重启运行：**

![在这里插入图片描述][2a3ce701ac7c41cb82b62c6b4bc17d71.png]  
发现此时便符合我们的预期了。

## 四、发布（Publish）和订阅（Subscribe） ##

在消息队列中，发布订阅模型是一种广泛应用的消息传递模型，它允许将同一消息发送给多个消费者，实现消息的广播。在发布订阅模型中，引入了交换机（Exchange）的概念，它充当了消息的分发中心。

**常见的交换机类型有三种：**

*  **Fanout Exchange（广播交换机）：** 将消息广播到所有绑定到该交换机的队列，忽略路由键。
 *  **Direct Exchange（直连交换机）：** 将消息路由到与消息的路由键匹配的队列，是一种点对点的路由。
 *  **Topic Exchange（主题交换机）：** 将消息路由到与消息的路由键匹配的队列，支持通配符匹配。

![a2c1bcde02584cb1bddeb4cc6850e7c4.png][]

说明：

*  `publisher` 将消息发送给 `exchange`，然后由 `excahnge` 根据路由规则发送至指定的队列；
 *  队列收到了消息之后，再发给自己的消费者

注意：`exchange` 负责消息路由，而不复制对消息的存储，路由失败消息则会丢失。

## 五、实现 Fanout（广播）发布订阅模型 ##

### 5.1 发布订阅 - Fanout Exchange ###

Fanout Exchange 是一种发布订阅模型，它会将接收到的消息广播到所有与其绑定的队列。每个队列都会收到相同的消息，从而实现消息的广播。

![Fanout Exchange][]

实现思路如下：

1.  在 `consumer` 服务中，利用代码声明队列、交换机，并将两者绑定。
2.  在 `consumer` 服务中，编写两个消费者方法，分别监听 `fanout.queue1` 和 `fanout.queue2`。
3.  在 `publisher` 中编写测试方法，向 `demo.fanout` 发送消息。

### 5.2 消息发布者 ###

在发布者模块中，我们创建了一个测试方法 `testSendFanoutExchange`，用于向 Fanout Exchange 发送消息。以下是该方法的代码和说明：

/**
     * 发生消息到交换机 demo.fanout
     */
    @Test
    public void testSendFanoutExchange(){
        
        // 交换机名称
        String exchangeName = "demo.fanout";
        // 信息
        String message = "hello every one!";
    
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

**说明：**

*  `exchangeName`：指定要发送消息的交换机名称，这里是 `demo.fanout`，即 Fanout Exchange 的名称。
 *  `message`：要发送的消息内容。
 *  `rabbitTemplate.convertAndSend(exchangeName, "", message)`：通过 RabbitTemplate 发送消息到指定的交换机，第二个参数是路由键，对于 Fanout Exchange，路由键为空字符串，因为 Fanout Exchange 会将消息广播到所有与之绑定的队列。

### 5.3 消息订阅者 ###

Spring AMQP 提供了声明交换机、队列、绑定关系的API，例如：

![03831c210d6f4678be552b70b9d88c79.png][]  
通过上述继承关系了可以发现，Fanout 模式需要使用的就是 `FanoutExchange` 类。

在订阅者模块中，创建了一个配置类 `FanoutConfig`，声明了 `Fanout Exchange`、两个队列，并将这两个队列分别绑定到 `Fanout Exchange`。

@Configuration
    public class FanoutConfig {
        
    
        // 声明 Fanout Exchange
        @Bean
        public FanoutExchange fanoutExchange() {
        
            return new FanoutExchange("demo.fanout");
        }
    
        // 声明队列 fanout.queue1
        @Bean
        public Queue fanoutQueue1() {
        
            return new Queue("fanout.queue1");
        }
    
        // 绑定队列 fanoutQueue1 到交换机 demo.fanout
        @Bean
        public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        
            return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
        }
    
        // 声明队列 fanout.queue2
        @Bean
        public Queue fanoutQueue2() {
        
            return new Queue("fanout.queue2");
        }
    
        // 绑定队列 fanoutQueue2 到交换机 demo.fanout
        @Bean
        public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        
            return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
        }
    
        @Bean
        public Queue objectQueue(){
        
            return new Queue("object.queue");
        }
    }

然后，编写了两个监听方法 `listenFanoutQueue1` 和 `listenFanoutQueue2`，分别用于监听两个队列，接收 Fanout Exchange 广播的消息。

@RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String message) {
        
        System.out.println("消费者收到 fanout.queue1 的消息：【" + message + "】");
    }
    
    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String message) {
        
        System.out.println("消费者收到 fanout.queue2 的消息：【" + message + "】");
    }

**说明：**

*  `FanoutConfig` 配置类中声明了 Fanout Exchange（`demo.fanout`）和两个队列（`fanout.queue1` 和 `fanout.queue2`），并通过 `Binding` 将队列和交换机进行了绑定。
 *  `@RabbitListener` 注解用于声明监听方法，其中 `queues` 参数指定监听的队列名称。
 *  当 Fanout Exchange 收到消息时，会将消息广播给所有与之绑定的队列，因此两个监听方法会同时收到相同的消息。

这样，我们就完成了 Fanout 发布订阅模型的实现，实现了消息的广播效果。

**效果演示：**

![60c0fe1cd53041388192af7494f53d9e.png][]

## 六、实现 Direct（路由）发布订阅模型 ##

### 6.1 发布订阅 - Direct Exchange ###

Direct Exchange 是一种路由模式，根据消息的 Routing Key 将消息路由到与之绑定的队列。在 Direct Exchange 中，每个队列与 Exchange 绑定时都需要指定一个 Binding Key，消息的 Routing Key 与 Binding Key 一致时，消息会被路由到对应的队列。

**以下是 Direct Exchange 模型的示意图：**

![aa2041442aab465cad23dd2bddef3e96.png][]

**模型说明：**

*  Exchange（DirectExchange）：Direct Exchange 负责消息的路由，根据消息的 Routing Key 将消息发送到对应的队列。
 *  Queue1、Queue2：两个队列，分别与 Exchange 绑定，指定了各自的 Binding Key。
 *  Routing Key：消息的路由键，由生产者指定。Exchange 根据 Routing Key 将消息发送到与之绑定的队列。

**示例说明：**

例如，上图中：

*  `Queue1` 与 `DirectExchange` 绑定时指定了两个 Binding Key：`red` 和 `blue`。
 *  `Queue2` 与 `DirectExchange` 绑定时指定了两个 Binding Key：`red` 和 `yellow`。

当生产者发送消息时，指定了消息的 Routing Key，Exchange 根据 Routing Key 将消息路由到与之匹配的队列。例如，如果 Routing Key 为 `red`，则消息将被路由到 `Queue1` 和 `Queue2`。

接下来，我们将实现一个基于 Direct Exchange 的发布订阅模型，包括生产者发送消息和两个消费者分别监听不同的队列。

### 6.2 消息发布者 ###

在消息发布者中，我们通过 `rabbitTemplate` 发送消息到 Direct Exchange。为了模拟不同的消息，我们发送了三条消息，分别指定了不同的 Routing Key。这样，消息就会被路由到与之匹配的队列。以下是对上述代码的说明：

/**
     * 发送消息到交换机 demo.direct
     */
    @Test
    public void testSendDirectExchange(){
        
        // 交换机名称
        String exchangeName = "demo.direct";
        // 信息
        String message1 = "hello, blue!";
        String message2 = "hello, yellow!";
        String message3 = "hello, red!";
    
        // 发送消息到 Exchange，并指定 Routing Key
        rabbitTemplate.convertAndSend(exchangeName, "blue", message1);
        rabbitTemplate.convertAndSend(exchangeName, "yellow", message2);
        rabbitTemplate.convertAndSend(exchangeName, "red", message3);
    }

说明：

*  `exchangeName`：交换机的名称，这里为 `demo.direct`。
 *  `message1`、`message2`、`message3`：三条不同的消息。
 *  使用 `rabbitTemplate.convertAndSend` 方法发送消息到 Direct Exchange，同时指定了消息的 Routing Key。消息的 Routing Key 决定了消息将被路由到哪个队列。

### 6.3 消息订阅者 ###

在消息订阅者中，我们使用 `@RabbitListener` 注解来绑定队列、交换机、以及绑定键（Binding Key）。这样，我们就可以通过注解的方式简便地声明消息的接收规则。

// 直接使用 @RabbitListener 注解绑定 队列、交换机、bindKey
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "demo.direct", type = ExchangeTypes.DIRECT),
            key = {
        "red", "blue"}
    ))
    public void listenDirectQueue1(String message) {
        
        System.out.println("消费者1 收到 direct.queue1 的消息：【" + message + "】");
    }
    
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "demo.direct", type = ExchangeTypes.DIRECT),
            key = {
        "red", "yellow"}
    ))
    public void listenDirectQueue2(String message) {
        
        System.out.println("消费者2 收到 direct.queue2 的消息：【" + message + "】");
    }

说明：

*  `@RabbitListener(bindings = @QueueBinding(...))`：使用此注解来声明队列的绑定关系。
 *  `@Queue(name = "direct.queue1")`：声明队列的名称为 `direct.queue1`。
 *  `@Exchange(name = "demo.direct", type = ExchangeTypes.DIRECT)`：声明交换机的名称为 `demo.direct`，类型为 `DIRECT`。
 *  `key = {"red", "blue"}`：指定队列与交换机之间的绑定键（Binding Key）。队列 `direct.queue1` 会接收 Routing Key 为 `red` 或 `blue` 的消息。

**接下来，我们可以运行测试，观察消息的发布和订阅过程：**  
![ea5cd1bf84474ca796460459d6b53127.png][]

通过运行结果可以发现：

*  由于 `queue1` 和 `queue2` 都绑定了 `red` 关键字，因此两个消费者都能够收到 `RoutingKey` 为 `red` 的消息。基于这个特点，路由（Direct）也可以实现广播（Fanout）的功能；

## 七、实现 Topic（广播）发布订阅模型 ##

### 7.1 发布订阅 - Topic Exchange ###

在消息队列系统中，发布订阅模式是一种强大的消息传递范式，而Topic Exchange则为这一模式提供了更灵活的消息路由机制。相较于Direct Exchange，Topic Exchange的routing key可以是多个单词的列表，通过`.`进行分割，队列可以使用通配符来指定关注的主题：

*  `#`：代指 0 个或多个单词；
 *  `*`：代指一个单词。

**例如，现在由如下关键字：**

*  china.news 代表有中国的新闻消息；
 *  china.weather 代表中国的天气消息；
 *  usa.news 则代表美国新闻
 *  usa.weather 代表美国的天气消息；

![71f3ef24fcac4a9da5173596eb14fe91.png][]

**对上述图示的说明：**

*  `queue1` 绑定关键字 `china.#`，表示该队列关注中国的所有消息。
 *  `queue2` 绑定关键字 `usa.#`，表示该队列关注美国的所有消息。
 *  `queue3` 绑定关键字 `#.weather`，表示该队列关注所有与天气有关的消息。
 *  `queue4` 绑定关键字 `#.news`，表示该队列关注所有与新闻有关的消息。

**根据上述绑定关系，以下是一些消息的匹配规则：**

*  `china.news` 匹配 `queue1` 的绑定，因为它以 `china.#` 开头，被发送到 `queue1`。
 *  `china.weather` 同样匹配 `queue1` 的绑定，因为它以 `china.#` 开头，也会被发送到 `queue1`。
 *  `usa.news` 匹配 `queue2` 的绑定，因为它以 `usa.#` 开头，被发送到 `queue2`。
 *  `usa.weather` 同样匹配 `queue2` 的绑定，因为它以 `usa.#` 开头，也会被发送到 `queue2`。

### 7.2 消息发布者 ###

在消息发布者中，我们通过 `rabbitTemplate` 发送消息到 Topic Exchange。为了模拟不同的消息，发送了三条消息，分别指定了不同的 Routing Key。这样，消息就会被路由到与之匹配的队列。

/**
         * 发送消息到交换机 demo.topic
         */
        @Test
        public void testSendTopicExchange(){
        
            // 交换机名称
            String exchangeName = "demo.topic";
            // 信息
            String message1 = "中国新闻！";
            String message2 = "美国新闻";
            String message3 = "中国天气";
    
            rabbitTemplate.convertAndSend(exchangeName, "china.news", message1);
            rabbitTemplate.convertAndSend(exchangeName, "usa.news", message2);
            rabbitTemplate.convertAndSend(exchangeName, "china.weather", message3);
        }

**说明：**

*  `testSendTopicExchange` 方法用于测试向 Topic Exchange 发送消息。
 *  交换机名称为 “demo.topic”。
 *  通过 `convertAndSend` 方法分别发送了关于中国新闻、美国新闻和中国天气的消息，指定了相应的 RoutingKey。

### 7.3 消息订阅者 ###

/**
     * 消费者监听队列 topic.queue1
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue1"),
            exchange = @Exchange(name = "demo.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String message) {
        
        System.out.println("消费者收到 topic.queue1 的消息：【" + message + "】");
    }
    
    /**
     * 消费者监听队列 topic.queue2
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("topic.queue2"),
            exchange = @Exchange(name = "demo.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String message) {
        
        System.out.println("消费者收到 topic.queue2 的消息：【" + message + "】");
    }

**说明：**

*  `listenTopicQueue1` 方法通过 `@RabbitListener` 注解监听队列 “topic.queue1”。
 *  `listenTopicQueue2` 方法通过 `@RabbitListener` 注解监听队列 “topic.queue2”。
 *  使用 `@QueueBinding` 注解进行队列、交换机、RoutingKey 的绑定。

通过上述配置，消息发布者向 Topic Exchange 发送消息时，根据指定的 RoutingKey，消息会被路由到相应的队列，实现了灵活的消息订阅和分发。

**测试结果：**

![347d7c630c314ae0b2b7f05e5923ae61.png][]

## 八、AMQP 的消息转换器 ##

### 8.1 测试发送 Object 类型消息 ###

在 Spring AMQP 的发送方法中，接收消息的类型是 `Object`，也就是说我们可以发送任意对象类型的消息，Spring AMQP 会帮我们序列化为字节后发送。

在 `publisher` 服务中编写发送消息的测试代码：

@Test
        public void testSendObjectQueue(){
        
            Map<String, Object> message = new HashMap<>();
            message.put("name", "张三");
            message.put("age", 18);
            rabbitTemplate.convertAndSend("object.queue", message);
        }

**说明：**

*  `testSendObjectQueue` 方法用于测试向 `object.queue` 队列发送消息。
 *  创建一个 `HashMap` 对象，设置了姓名和年龄，并将其发送到队列。

**然后发送消息：**

![8866408a78c84107a54b7913930569e4.png][]

此时发现，在消息队列中的类型变成了 `application/x-java-serialized-object`，原因是默认采用的序列化方法为 Java 的默认序列化。

### 8.2 消息转换器 ###

在 Spring AMQP 中，消息对象的处理由 `org.springframework.amqp.support.converter.MessageConverter` 负责，而默认的实现是 `SimpleMessageConverter`，它基于 JDK 的 `ObjectOutputStream` 完成序列化。这是为了提供通用性，但在某些场景下，我们可能需要自定义消息的序列化方式。

**Spring AMQP 提供了多个消息转换器，其中包括：**

1.  **SimpleMessageConverter：** 默认的消息转换器，基于 JDK 的 `ObjectOutputStream` 进行序列化。
2.  **Jackson2JsonMessageConverter：** 使用 Jackson 库将对象转换为 JSON 格式，适合于处理 JSON 数据。
3.  **MarshallingMessageConverter：** 使用 Spring 的 Marshaller 进行序列化，可以支持 XML 等格式。
4.  **SmartMessageConverter：** 一个更加灵活的接口，允许自定义消息转换器。

**自定义消息转换器：**

要自定义消息转换器，我们可以实现 `MessageConverter` 接口，然后在配置类中注册为 Bean。以下是一个简单的例子：

import org.springframework.amqp.core.Message;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.stereotype.Component;
    
    @Component
    public class CustomMessageConverter implements MessageConverter {
        
    
        @Override
        public Message toMessage(Object object, MessageProperties messageProperties) {
        
            // 实现将对象转换为 Message 的逻辑
            // ...
            return null;
        }
    
        @Override
        public Object fromMessage(Message message) {
        
            // 实现将 Message 转换为对象的逻辑
            // ...
            return null;
        }
    }

在上述例子中，`CustomMessageConverter` 实现了 `MessageConverter` 接口，并重写了 `toMessage` 和 `fromMessage` 方法，用于实现自定义的消息转换逻辑。

然后，在配置类中将该转换器注册为 Bean：

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMQConfig {
        
    
        @Bean
        public MessageConverter customMessageConverter() {
        
            return new CustomMessageConverter();
        }
    
        // 其他配置...
    }

这样，我们就可以在应用中使用自定义的消息转换器了。

### 8.3 更改消息转换器类型 ###

如果希望使用 JSON 格式来序列化对象，可以通过使用 `Jackson2JsonMessageConverter` 来实现。以下是具体步骤：

1.  添加 Jackson 依赖

首先，在父工程的 `pom.xml` 中添加 Jackson 的依赖：

<dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>

1.  声明 Jackson2JsonMessageConverter

在 `publisher` 服务的配置类中，声明消息转换器的类型为 `Jackson2JsonMessageConverter`：

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.messaging.converter.MessageConverter;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    
    @Configuration
    public class RabbitMQConfig {
        
    
        @Bean
        public MessageConverter messageConverter(){
        
            return new Jackson2JsonMessageConverter();
        }
    
        // 其他配置...
    }

上述代码通过 `@Bean` 注解创建了一个 `Jackson2JsonMessageConverter` 的 Bean，并将其注册为消息转换器。这样，消息将以 JSON 格式进行序列化。

1.  测试发送 Object 类型消息

然后，在 `publisher` 服务中编写发送消息的测试代码：

@Test
    public void testSendObjectQueue(){
        
        Map<String, Object> message = new HashMap<>();
        message.put("name", "张三");
        message.put("age", 18);
        rabbitTemplate.convertAndSend("object.queue", message);
    }

运行测试代码，发现此时的消息类型变成了`application/json`，此时消息的占用空间更少，也更加利于阅读。

![5b679b5bfd5a43a1ba9799a648aa243f.png][]

通过这样的配置，消息将以 JSON 格式进行序列化，而不再是默认的 Java 序列化方式。这样可以更灵活地处理不同格式的消息。

[https_spring.io_projects_spring-amqp]: https://spring.io/projects/spring-amqp
[b1465671d4ce4a329bd53df9fb872cd6.png]: https://img-blog.csdnimg.cn/b1465671d4ce4a329bd53df9fb872cd6.png
[8cab6f5bcf084d388cd00d33f2b1592f.png]: https://img-blog.csdnimg.cn/8cab6f5bcf084d388cd00d33f2b1592f.png
[6636463a4140467f985adecdaf3f6139.png]: https://img-blog.csdnimg.cn/6636463a4140467f985adecdaf3f6139.png
[86b36d59efb14f268c8f5acc437a9d9b.png]: https://img-blog.csdnimg.cn/86b36d59efb14f268c8f5acc437a9d9b.png
[8890cbd767df4f92a45ef564e3502df9.png]: https://img-blog.csdnimg.cn/8890cbd767df4f92a45ef564e3502df9.png
[2a3ce701ac7c41cb82b62c6b4bc17d71.png]: https://img-blog.csdnimg.cn/2a3ce701ac7c41cb82b62c6b4bc17d71.png
[a2c1bcde02584cb1bddeb4cc6850e7c4.png]: https://img-blog.csdnimg.cn/a2c1bcde02584cb1bddeb4cc6850e7c4.png
[Fanout Exchange]: https://img-blog.csdnimg.cn/649a0aac7a854ec2af52e5e673087c80.png
[03831c210d6f4678be552b70b9d88c79.png]: https://img-blog.csdnimg.cn/03831c210d6f4678be552b70b9d88c79.png
[60c0fe1cd53041388192af7494f53d9e.png]: https://img-blog.csdnimg.cn/60c0fe1cd53041388192af7494f53d9e.png
[aa2041442aab465cad23dd2bddef3e96.png]: https://img-blog.csdnimg.cn/aa2041442aab465cad23dd2bddef3e96.png
[ea5cd1bf84474ca796460459d6b53127.png]: https://img-blog.csdnimg.cn/ea5cd1bf84474ca796460459d6b53127.png
[71f3ef24fcac4a9da5173596eb14fe91.png]: https://img-blog.csdnimg.cn/71f3ef24fcac4a9da5173596eb14fe91.png
[347d7c630c314ae0b2b7f05e5923ae61.png]: https://img-blog.csdnimg.cn/347d7c630c314ae0b2b7f05e5923ae61.png
[8866408a78c84107a54b7913930569e4.png]: https://img-blog.csdnimg.cn/8866408a78c84107a54b7913930569e4.png
[5b679b5bfd5a43a1ba9799a648aa243f.png]: https://img-blog.csdnimg.cn/5b679b5bfd5a43a1ba9799a648aa243f.png