RabbitMQ：Fanout、Direct、Topic 交换机、队列声明与消息转换器

承接上一篇《RabbitMQ：AMQP 原理、Spring AMQP 实战与 Work Queue 模型》，本文将深入实践 RabbitMQ 中最常用的三种交换机——Fanout、Direct 与 Topic，通过 Spring AMQP 和控制台操作，直观理解它们的消息分发规则、消费行为和适用场景，同时展示标准的交换机、队列与绑定声明方法及 JSON 消息转换实践

一、Fanout 交换机

（一）实验目的

通过 Spring AMQP 实践 RabbitMQ 中的 Fanout（广播）交换机，理解其消息分发机制：

一条消息会被复制并发送到所有绑定到该交换机的队列中。

（二）实现思路

整体流程如下：

（三）控制台操作

1. 声明两个队列

路径：Queues → Add a new queue

创建以下队列：

fanout.queue1
fanout.queue2

队列类型：Classic
持久化（Durable）：true

2. 声明 Fanout 交换机并绑定队列

在 Exchanges → Add a new exchange

创建交换机：

Name: c.fanout
Type: fanout

创建完成后，进入 c.fanout 交换机详情页：

在 Bindings → Add binding 中，依次添加绑定：

（1）绑定 fanout.queue1

（2）绑定 fanout.queue2

Fanout 交换机不依赖 routingKey，因此绑定时 routingKey 可留空。

（四）consumer 服务（两个消费者）

1. 编写消费者代码

@Slf4j
@Component
public class FanoutConsumer {

/**
* 消费者 1
*/
@RabbitListener(queues = "fanout.queue1")
public void listenQueue1(String msg) {
System.out.println("消费者1收到了【fanout.queue1】的消息：" + msg);
}

/**
* 消费者 2
*/
@RabbitListener(queues = "fanout.queue2")
public void listenQueue2(String msg) {
System.out.println("消费者2收到了【fanout.queue2】的消息：" + msg);
}
}

2. 关键说明

使用的是 两个不同的队列

每个队列对应一个独立的消费者

消费过程 互不竞争、互不影响

（五）publisher 服务（发送广播消息）

1. 发送消息代码

@Test
void testFanoutSend() {
String msg = "Fanout 广播消息";

rabbitTemplate.convertAndSend(
"c.fanout", // 交换机
"", // routingKey，fanout 会忽略
msg
);
System.out.println("消息已发送：" + msg);
}

2. 说明

消息直接发送到 交换机，而不是队列

routingKey 对 Fanout 交换机 不起作用

（六）运行效果

1. 启动 consumer 服务

2. 执行 publisher 中的测试方法

控制台输出：

消费者1收到了【fanout.queue1】的消息：Fanout 广播消息
消费者2收到了【fanout.queue2】的消息：Fanout 广播消息

可以看到：

同一条消息，被两个队列分别完整消费了一次。

（七）总结：交换机的作用是什么？

RabbitMQ 中，交换机（Exchange）承担着消息路由的核心职责：

1. 接收生产者（Publisher）发送的消息

2. 根据交换机类型和绑定规则，将消息路由到一个或多个队列

FanoutExchange 的特点总结

不关心 routingKey

将消息 广播 给所有与之绑定的队列

每个队列都会收到一份完整的消息副本

FanoutExchange = 广播模型（多队列复制）

二、Direct交换机

Direct Exchange 会将接收到的消息按照 RoutingKey 的精确匹配规则路由到指定的 Queue，因此也称为定向路由。

每个 Queue 与 Exchange 通过 BindingKey 建立绑定关系

发布者发送消息时，指定 RoutingKey

Exchange 仅会将消息投递到 BindingKey 与 RoutingKey 完全一致的队列

案例：

使用 Spring AMQP 演示 DirectExchange 的基本用法。

需求

1. 在 RabbitMQ 控制台中声明两个队列：direct.queue1，direct.queue2

2. 声明一个 Direct 类型交换机：c.direct，并将两个队列绑定到该交换机

3. 在 consumer 服务中编写两个消费者方法，分别监听上述两个队列

4. 在 publisher 服务中，使用不同的 RoutingKey 向 c.direct 发送消息

（一）Direct 交换机规则

消息的 RoutingKey 必须与队列绑定时的 RoutingKey 完全一致，消息才会被路由到该队列。

        精确匹配（不支持通配符）

        不匹配的消息将被直接丢弃（若未配置备份交换机）

（二）RabbitMQ 控制台操作

1. 声明队列

在 RabbitMQ 控制台中创建两个队列：

direct.queue1
direct.queue2

队列类型：Classic
持久化（Durable）：true

2. 声明 Direct 交换机

创建交换机：

Name：c.direct
Type：direct
Durable：true

3. 绑定关系（关键）

注意：
Direct 模式下，RoutingKey 必须一字不差，否则消息无法被路由。

（三）consumer 服务

1. 消费者代码

@Slf4j
@Component
public class DirectConsumer {

@RabbitListener(queues = "direct.queue1")
public void listenQueue1(String msg) {
System.out.println("消费者1收到了【direct.queue1】的消息：" + msg);
}

@RabbitListener(queues = "direct.queue2")
public void listenQueue2(String msg) {
System.out.println("消费者2收到了【direct.queue2】的消息：" + msg);
}
}

（四）publisher 服务

1. 测试发送消息

@Test
void testDirectExchange() {
// routingKey = red → queue1
rabbitTemplate.convertAndSend(
"c.direct",
"red",
"这是一条 red 消息"
);

// routingKey = blue → queue2
rabbitTemplate.convertAndSend(
"c.direct",
"blue",
"这是一条 blue 消息"
);

// routingKey = green → 没有队列接收
rabbitTemplate.convertAndSend(
"c.direct",
"green",
"这是一条 green 消息"
);
}

（五）运行结果（现象验证）

1. consumer 控制台输出

消费者2收到了【direct.queue2】的消息：这是一条 blue 消息
消费者1收到了【direct.queue1】的消息：这是一条 red 消息

说明：

red 消息被成功路由到 direct.queue1

blue 消息被成功路由到 direct.queue2

green 消息因无匹配的 BindingKey，被直接丢弃

总结

DirectExchange 采用 精确路由 机制

RoutingKey 与 BindingKey 必须完全一致

适用于 日志级别区分、消息类型分发、业务指令路由 等场景

至此，一个完整的 Direct 交换机实践示例就完成了。

三、Topic 交换机

TopicExchange 与 DirectExchange 的核心区别：

routingKey 是 多个单词，用 . 分隔
例如：order.create, log.error.db

绑定时（BindingKey）可以使用通配符

        * ：匹配 一个单词

        # ：匹配 0 个或多个单词

（一）案例目标

利用 SpringAMQP 演示 TopicExchange

1. RabbitMQ 控制台

声明队列：topic.queue1，topic.queue2

声明交换机：c.topic（类型：topic）

绑定关系：topic.queue1 ← order.*，topic.queue2 ← order.#

2. consumer 服务

两个消费者方法,分别监听 topic.queue1、topic.queue2

3. publisher 服务

向 c.topic 发送不同 routingKey 的消息，观察路由效果

（二）RabbitMQ 控制台操作

1. 创建 Topic 交换机

Exchanges → Add Exchange

Name：c.topic

Type：topic

Durability：Durable

点击 Add exchange

2. 创建队列

创建两个队列：topic.queue1，topic.queue2，参数保持默认即可。

3. 绑定队列与交换机

进入 c.topic → Bindings → Add binding

（三）Consumer 服务（消费者）

1. 编写消费者类

@Slf4j
@Component
public class TopicConsumer {

@RabbitListener(queues = "topic.queue1")
public void listenQueue1(String msg) {
System.out.println("消费者1收到了【topic.queue1】的消息：" + msg);
}

@RabbitListener(queues = "topic.queue2")
public void listenQueue2(String msg) {
System.out.println("消费者2收到了【topic.queue2】的消息：" + msg);
}
}

注：消费者只需监听队列，队列与交换机的绑定已经在控制台完成。

（四）Publisher 服务（生产者）

1. 发送消息的测试代码

@Test
void testTopicExchange() {

// routingKey：order.create
rabbitTemplate.convertAndSend(
"c.topic",
"order.create",
"订单创建消息"
);

// routingKey：order.pay.success
rabbitTemplate.convertAndSend(
"c.topic",
"order.pay.success",
"订单支付成功消息"
);

// routingKey：order
rabbitTemplate.convertAndSend(
"c.topic",
"order",
"订单根级消息"
);
}

（五）运行结果

1. consumer 控制台输出

消费者1收到了【topic.queue1】的消息：订单创建消息
消费者2收到了【topic.queue2】的消息：订单创建消息
消费者2收到了【topic.queue2】的消息：订单支付成功消息
消费者2收到了【topic.queue2】的消息：订单根级消息

分析：

topic.queue1 绑定 order.*

只能匹配 两个单词，所以只收到 order.create 消息

topic.queue2 绑定 order.#

匹配 0 或多个单词，所以收到所有以 order 开头的消息

四、声明队列 / 交换机 / 绑定关系 ——标准实践方式

目标：
不发送任何消息，仅通过 Spring Boot 启动完成 MQ 结构声明（交换机、队列、绑定），支持两种方式：配置类方式 & 注解方式

（一）方式一：配置类方式

1. 配置类：MqDeclareConfig.java

@Configuration
public class MqDeclareConfig {

// 1. Fanout 交换机
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("demo.fanout");
}

// 2️. 第一个队列
@Bean
public Queue fanoutQueueA() {
return new Queue("fanout.queueA");
}

// 3️. 第二个队列
@Bean
public Queue fanoutQueueB() {
return new Queue("fanout.queueB");
}

// 4. 绑定 queueA
@Bean
public Binding bindQueueA(
Queue fanoutQueueA,
FanoutExchange fanoutExchange) {

return BindingBuilder
.bind(fanoutQueueA)
.to(fanoutExchange);
}

// 5️. 绑定 queueB
@Bean
public Binding bindQueueB(
Queue fanoutQueueB,
FanoutExchange fanoutExchange) {

return BindingBuilder
.bind(fanoutQueueB)
.to(fanoutExchange);
}
}

说明

Spring Boot 启动时自动声明交换机、队列、绑定关系

不需要手动在控制台创建 MQ 结构

可以集中管理 MQ 架构，适合复杂场景

（二）方式二：基于注解声明队列/交换机/绑定

消费者类示例：FanoutConsumer.java

@Component
public class FanoutConsumer1 {

@RabbitListener(
bindings = @QueueBinding(
value = @Queue(name = "fanout.queueC", durable = "true"),
exchange = @Exchange(name = "demo1.fanout", type = ExchangeTypes.FANOUT)
)
)
public void receiveC(String message) {
System.out.println("fanout.queueC 收到消息：" + message);
}

@RabbitListener(
bindings = @QueueBinding(
value = @Queue(name = "fanout.queueD", durable = "true"),
exchange = @Exchange(name = "demo1.fanout", type = ExchangeTypes.FANOUT)
)
)
public void receiveD(String message) {
System.out.println("fanout.queueD 收到消息：" + message);
}
}

说明：

@QueueBinding 同时声明队列、交换机和绑定

队列声明与消费者绑定在一起，代码更简洁

Spring Boot 启动时自动声明 MQ 结构

如果队列或交换机已存在，自动忽略，不报错

（三）启动项目验证

1. 操作步骤

（1）启动 Spring Boot 项目

（2）打开 RabbitMQ 控制台

（3）查看 MQ 结构是否存在：

Exchanges → demo.fanout、demo1.fanout

Queues → fanout.queueA、fanout.queueB、fanout.queueC、fanout.queueD

Bindings 是否存在

此时无需发送任何消息，结构已经声明成功

这一步验证的是：Spring 声明能力，而不是消息发送

2. 总结

两种方式可以共存，也可以单独使用。
如果项目中队列比较少，注解方式更简洁；如果 MQ 架构复杂或需要统一管理，配置类方式更合适。

五、消息转换器—— 对象消息 JSON 化实践

目标：
理解为什么默认是 JDK 序列化，以及如何切换为 JSON

（一）声明一个“对象消息队列”

MqDeclareConfig.java 中追加

@Bean
public Queue objectQueue() {
return new Queue("object.queue");
}

（二）发送对象消息（默认：错误示范）

ObjectPublisher.java

@Component
public class ObjectPublisher {

@Autowired
private RabbitTemplate rabbitTemplate;

public void send() {
Map<String, Object> msg = new HashMap<>();
msg.put("name", "xiaoman");
msg.put("age", 18);

rabbitTemplate.convertAndSend("object.queue", msg);
}
}

（三）测试类发送消息

SendObjectTest.java

@SpringBootTest
public class SendObjectTest {

@Autowired
private ObjectPublisher publisher;

@Test
void testSendObject() {
publisher.send();
}
}

（四）查看 RabbitMQ 控制台（验证默认行为）

打开 object.queue，可以看到：

消息体是 乱码

Content-Type：

application/x-java-serialized-object

结论

默认 MessageConverter = SimpleMessageConverter
使用 JDK 序列化

这是错误示范，但非常重要 —— 它证明了不配置 JSON，Spring 默认就是 JDK 序列化

（五）引入 JSON 消息转换器（正确做法）

1. 添加依赖

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

2. 配置 MessageConverter

MessageConverterConfig.java

@Configuration
public class MessageConverterConfig {

@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}

（六）为什么 MessageConverter 必须生产者和消费者都写？

这是最核心的一点。

（1）在 publisher（生产者）侧作用

作用：
 Java 对象 → JSON 字节数组

流程：

Map / Object
↓
Jackson2JsonMessageConverter
↓
JSON
↓
RabbitMQ

如果 publisher 不配置：

使用默认 SimpleMessageConverter

消息变成 JDK 序列化

控制台显示乱码

（2）在 consumer（消费者）侧作用

作用：
JSON → Java 对象

流程：

RabbitMQ
↓
JSON
↓
Jackson2JsonMessageConverter
↓
Map / Object

如果 consumer 不配置：

Spring 无法反序列化 JSON

可能报错

或只能拿到 byte[] / String

序列化在发送端，反序列化在接收端
两端都需要 MessageConverter

（七）再次发送消息，验证成功

重新运行测试

RabbitMQ 控制台显示：

{"name":"xiaoman","age":18}

可读
标准 JSON
正确 Content-Type：application/json

（八）消费者接收对象消息

ObjectConsumer.java

@Component
public class ObjectConsumer {

@RabbitListener(queues = "object.queue")
public void listen(Map<String, Object> msg) {
System.out.println("name = " + msg.get("name"));
System.out.println("age = " + msg.get("age"));
}
}

控制台输出：

name = xiaoman
age = 18

六、总结

        通过本文的实践，可以全面理解 RabbitMQ 三种核心交换机的机制与应用：Fanout 负责广播消息到所有绑定队列，适合事件通知和配置刷新场景；Direct 实现精确路由，消息仅发送到匹配 RoutingKey 的队列，适合日志分级、业务指令分发；Topic 支持多单词路由键及通配符匹配，兼顾灵活性与可控性，适用于订单、日志和复杂事件路由。同时，实践也验证了工程化最佳实践的重要性：通过 Spring AMQP 声明交换机、队列及绑定关系，可以保证环境一致性并减少人工操作错误；而在生产者和消费者双方统一配置 JSON MessageConverter，则确保对象消息格式正确、可读且跨语言兼容。这些经验不仅帮助开发者准确理解消息路由机制，也为构建稳定、高可维护性的分布式消息系统奠定了基础。