RabbitMQ 四种交换机（Direct、Topic、Fanout、Headers）详解

---

本文是博主在梳理 RabbitMQ 知识的过程中，将所遇到和可能会遇到的基础知识记录下来，用作梳理 RabbitMQ 的整体架构和功能的线索文章，通过查找对应的知识能够快速的了解对应的知识而解决相应的问题。

---

RabbitMQ 是一个开源的消息中间件（Message Broker），基于 AMQP（高级消息队列协议，Advanced Message Queuing Protocol）实现，旨在为分布式系统提供可靠的异步通信解决方案。

在开发中引入 RabbitMQ 中间件一般用于不同项目之间的消息传递，利用其基本特性可以应对各种常见问题。

• 异步通信桥梁：允许不同应用程序通过 消息 进行间接通信，无需实时同步响应，提升系统解耦性和可扩展性。
• 削峰填谷：通过消息队列暂存峰值流量，缓解后端服务压力，适用于流量波动较大的场景（如电商促销、日志处理）。
• 可靠消息传递：支持消息持久化、确认机制（ACK）、重试策略等，确保消息不丢失或重复处理。
• 多语言支持：提供 Java、Python、C#、Go、JavaScript 等主流语言的客户端库，方便跨平台集成。
• 高可用性与集群：支持节点集群部署（如镜像队列），实现故障转移和负载均衡，保障服务稳定性。
• 插件生态：通过插件扩展功能，例如管理界面（RabbitMQ Management Plugin）、STOMP/MQTT 协议支持、消息追踪等。
• 轻量与高效：基于 Erlang 语言开发，天生支持高并发和低延迟，适合分布式系统环境。与 Kafka 等吞吐量优先的中间件相比，RabbitMQ 在处理超大规模数据时性能稍弱，更适合中等规模、注重可靠性的业务。

本文主要讲解 RabbitMQ 支持的四种类型交换机：

首先，RabbitMQ是一个消息代理，负责接收和转发消息。在RabbitMQ中，生产者发送消息到交换机，交换机根据类型和绑定规则将消息路由到队列。消费者从队列中获取消息进行处理。

这里的关键在于交换机的类型，不同的交换机类型决定了不同的消息路由方式。

一、直连交换机（Direct Exchange）

1、核心原理

直连交换机是 RabbitMQ 中最简单的消息路由机制，基于精确的路由键匹配规则。

直连交换机根据消息的路由键（Routing Key）和队列绑定时指定的绑定键（Binding Key）进行精确匹配。如果两者完全匹配，消息就会被路由到对应的队列。这种精确匹配的方式适用于需要精确控制消息路由的场景，比如根据不同的任务类型分发到不同的队列。

• 生产者发送消息时指定一个路由键（Routing Key）。
• 队列通过绑定键（Binding Key）与交换机绑定。
• 仅当路由键与绑定键完全匹配时，消息才会被路由到对应队列。

---

假设有一个直连交换机 direct_logs ，队列A绑定路由键 info，队列B绑定路由键 error。当生产者发送路由键为 info 的消息时，只有队列A会收到；发送 error 则只有队列B收到。如果发送 warning 且没有队列绑定该键，消息会被丢弃。

---

在RabbitMQ中，当声明一个队列时，如果没有指定交换机，会使用默认的直连交换机，队列名称作为路由键。

使用直连交换机要注意的一些特性：

• 精确匹配：路由键与绑定键需严格一致（如 order.create → order.create）。
• 多队列绑定：多个队列可绑定到同一路由键，消息会广播到所有匹配队列。
• 区分大小写：Error 与 error 视为不同键。
• 默认交换机：RabbitMQ 内置无名直连交换机，队列默认绑定到该交换机（路由键=队列名）。

常用的一些场景：

• 任务分类：例如按订单类型（payment、shipping）分发到不同队列。
• 日志级别：将 info、error 日志路由到不同的处理服务。

2、Java 代码示例

1、代码依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version>
</dependency>

2、生产者（发送消息）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class DirectExchangeProducer {private static final String EXCHANGE_NAME = "direct_logs";private static final String[] ROUTING_KEYS = {"info", "error", "warning"};private static final String QUEUE_NAME = "my-mq-queue";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);// 2. 建立连接并创建通道try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 3. 声明直连交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");// 4. 发送消息到不同路由键for (String routingKey : ROUTING_KEYS) {String message = "Log message of level: " + routingKey;channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");}}}
}

3、消费者（接收消息）

import com.rabbitmq.client.*;public class DirectExchangeConsumer {private static final String EXCHANGE_NAME = "direct_logs";private static final String QUEUE_NAME = "info_queue";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 1. 声明直连交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");// 2. 声明队列并绑定路由键channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info"); // 绑定键为 "info"// 3. 定义消息处理回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};// 4. 监听队列消费消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}
}

生产者发送消息：

• 发送路由键为 info → 投递到绑定键为 info 的队列。
• 发送路由键为 error → 投递到绑定键为 error 的队列。
• 发送路由键为 warning → 若无队列绑定该键，消息被丢弃。

消费者绑定队列：

• 队列 info_queue 绑定到 direct_logs 交换机，绑定键为 info。
• 仅接收路由键为 info 的消息。

4、注意事项

• 若无队列匹配路由键，消息会被直接丢弃。
• 多个消费者监听同一队列时，消息会在消费者间轮询分发（竞争消费模式）。
  • 多队列绑定同一键：例如同时绑定 queue1 和 queue2 到 error 键，消息会广播到两个队列。
• 直连交换机因匹配规则简单，路由性能极高，适合高吞吐场景。
• 结合其他交换机，例如先用 Topic Exchange 分类消息，再用 Direct Exchange 精确分发。

直连交换机通过精确的路由键匹配机制，为明确目标的消息分发提供了高效、直接的解决方案。

---

二、主题交换机（Topic Exchange）

1、核心原理

主题交换机是RabbitMQ中基于通配符模式匹配的灵活路由机制，与直连交换机不同的是，它可以支持模糊匹配机制。

• 生产者发送消息时指定一个路由键（Routing Key），格式为点分隔的单词（order.payment.success）。
• 队列通过绑定键（Binding Key）与交换机绑定，绑定键支持通配符（* 和 #）。
• 路由规则：根据绑定键的通配符模式匹配路由键，匹配成功则消息路由到对应队列。

---

通配符规则：

• *（星号）：
  • 匹配一个单词（任意字符组合）。
  • 示例：order.*.success → 匹配 order.payment.success，但不匹配 order.payment.email.success。
• #（井号）：
  • 匹配零个或多个单词（任意长度）。
  • 示例：order.# → 匹配 order.payment、order.payment.success、order 等。
• 绑定键格式：
  • 必须为点分隔的单词（如 user.notification.email）。
  • 通配符必须位于单词位置（如 *.error 是合法的，*error 非法）。

---

常用的一些场景：

• 多维度消息分类：例如按业务类型（order、payment）和操作结果（success、error）组合路由。
• 动态订阅：允许消费者根据自身需求绑定特定模式的路由键（如监听所有错误日志 *.error）。
• 层次化路由：支持树形结构的路由设计（如 region.us.east.notification）。

---

2、Java代码示例

1、代码依赖（Maven）

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version>
</dependency>

2、生产者（发送消息）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class TopicExchangeProducer {private static final String EXCHANGE_NAME = "topic_logs";private static final String[] ROUTING_KEYS = {"order.payment.success","order.shipping.error","user.notification.email","system.alert.high"};private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明主题交换机channel.exchangeDeclare(EXCHANGE_NAME, "topic");// 发送不同路由键的消息for (String routingKey : ROUTING_KEYS) {String message = "Message with routing key: " + routingKey;channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");}}}
}

3、消费者（接收消息）

import com.rabbitmq.client.*;public class TopicExchangeConsumer {private static final String EXCHANGE_NAME = "topic_logs";private static final String QUEUE_NAME = "order_errors";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明主题交换机channel.exchangeDeclare(EXCHANGE_NAME, "topic");// 声明队列并绑定路由键模式channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "order.*.error");// 定义消息处理回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};// 监听队列消费消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}
}

绑定键与路由键匹配：

• 队列A绑定 order.*.error → 匹配 order.shipping.error，不匹配 order.payment.success。
• 队列B绑定 #.error → 匹配 system.alert.error、order.shipping.error 等。
• 队列C绑定 system.# → 匹配 system.alert.high、system.monitoring.cpu 等。

消息分发：

• 路由键 order.shipping.error → 同时匹配队列A和队列C。
• 路由键 user.notification.email → 仅匹配绑定 user.*.email 或 user.# 或 #.email 的队列。

---

4、注意事项

• 通配符位置限制：
  • * 和 # 必须位于绑定键的单词位置，如 *.error 合法，error.* 或 err* 非法。
• 通配符匹配相比直连交换机的精确匹配略有性能损耗，但在大多数场景下可忽略。
• 若消息无匹配队列，默认会被丢弃。
• 可以组合使用交换机，例如先用 Fanout Exchange 广播消息，再用 Topic Exchange 精细化过滤。

主题交换机通过通配符模式匹配，提供了高度灵活的路由能力，适用于需要多维度分类、动态订阅或层次化消息分发的场景。

---

三、扇出交换机（Fanout Exchange）

1、核心原理

扇出交换机是RabbitMQ中最简单的广播型消息路由机制。

• 生产者发送消息到扇出交换机时，路由键（Routing Key）会被完全忽略。
• 所有绑定到该交换机的队列都会收到消息的副本，无论队列的绑定键是什么。
• 本质是“发布/订阅”模式：消息会被广播到所有关联的队列。

常见的特性：

• 无路由键匹配：路由键在发送时无效（可随意填写或为空）。
• 广播机制：消息会被复制并分发给所有绑定队列。
• 动态绑定：队列可在任何时刻绑定到交换机，实时生效。
• 性能高效：无需复杂匹配逻辑，适合高频消息分发。

常用于的场景：

• 实时通知系统：例如向所有在线用户广播系统公告。
• 日志聚合：将日志消息广播到多个处理服务（存储、分析、报警）。
• 事件驱动架构：多个服务需要同时响应同一事件（如订单创建事件）。
• 缓存同步：多个缓存节点需要同时接收数据更新消息。

---

2、Java代码示例

1、添加依赖（Maven）

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version>
</dependency>

2、生产者（发送广播消息）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class FanoutExchangeProducer {private static final String EXCHANGE_NAME = "fanout_notifications";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明扇出交换机（如果不存在则创建）channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 发送消息（路由键可任意填写，实际会被忽略）String message = "Broadcast message to all queues!";channel.basicPublish(EXCHANGE_NAME, "any_routing_key", // 此处路由键无效null, message.getBytes("UTF-8"));System.out.println(" [x] Sent: '" + message + "'");}}
}

3、消费者（接收广播消息）

import com.rabbitmq.client.*;public class FanoutExchangeConsumer {private static final String EXCHANGE_NAME = "fanout_notifications";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明扇出交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 创建临时队列（非持久化、独占、自动删除）String queueName = channel.queueDeclare().getQueue();// 将队列绑定到扇出交换机（绑定键无效，可填写空字符串）channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. Queue: " + queueName);// 定义消息处理回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received: '" + message + "'");};// 监听队列消费消息channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}
}

启动两个消费者：

• 消费者A创建队列 amq.gen-123 并绑定到 fanout_notifications。
• 消费者B创建队列 amq.gen-456 并绑定到同一交换机。

生产者发送消息：

• 消息发送到 fanout_notifications 交换机。
• 交换机会将消息复制并发送到 amq.gen-123 和 amq.gen-456 两个队列。

消费者接收结果：

• 消费者A和消费者B会同时收到相同的消息副本。

---

4、注意事项：

• 示例中使用临时队列（queueDeclare() 无参调用），实际生产环境需根据需求设置队列属性：

   // 持久化队列（服务器重启后保留）boolean durable = true;channel.queueDeclare("persistent_queue", durable, false, false, null);
  

• 示例中自动确认消息（autoAck=true），实际场景建议手动确认以保证可靠性：

  channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {});
  // 在消息处理完成后手动确认
  channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  

• 广播机制会导致消息被复制多份，需注意网络带宽压力（高并发场景需评估）和消费者处理能力（避免广播风暴）。

---

5、扩展使用：

• 组合其他交换机：先用 Fanout Exchange 广播消息，再用 Direct Exchange 精细化处理特定任务。
  • 例如：订单创建事件广播 → 库存服务扣减库存，日志服务记录日志，通知服务发送短信。
• 动态订阅系统：新服务上线时自动绑定到扇出交换机，立即接收所有消息。服务下线时解绑队列，停止接收消息。
• 多个缓存节点绑定到同一扇出交换机，接收数据更新消息以保持一致性。

扇出交换机通过简单粗暴的广播机制，为需要 一对多 消息分发的场景提供了高效解决方案。

---

四、头交换机（Headers Exchange）

1、核心原理

头交换机是RabbitMQ中基于消息头（Headers）键值对匹配的路由机制。

• 路由键（Routing Key）完全被忽略：消息的路由不依赖路由键，而是基于消息头的属性。
• 绑定规则：队列通过声明一组键值对条件与交换机绑定，消息头需满足这些条件才能路由到队列。
• 匹配模式：支持 x-match 参数指定匹配规则：
  • all：消息头需包含所有声明的键值对（逻辑与）。
  • any：消息头只需包含任意一个声明的键值对（逻辑或）。

核心特性：

• 不依赖路由键：路由键（如 basicPublish 中的 routingKey）在头交换机中无意义。
• 灵活匹配规则：支持复杂逻辑（all 或 any）匹配消息头的多个属性。
• 键值对匹配：匹配基于消息头的键值对，值可以是字符串或数值类型。
• 适用特殊场景：常用于需要多条件组合路由的场景，例如根据消息的元数据（如版本、环境）分发。

适用场景：

• 多条件路由：需要同时满足多个消息属性（如 type=order 且 priority=high）。
• 协议转换：消息头可携带协议元数据（如 format=json、version=2），路由到对应处理器。
• 环境隔离：根据 env=prod 或 env=test 将消息路由到不同环境队列。
• A/B测试：通过消息头标记用户分组（如 group=A），路由到不同实验队列。

---

2、Java代码示例

1、添加依赖（Maven）

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version>
</dependency>

2、生产者（发送消息）

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;public class HeadersExchangeProducer {private static final String EXCHANGE_NAME = "headers_logs";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明头交换机channel.exchangeDeclare(EXCHANGE_NAME, "headers");// 定义消息头Map<String, Object> headers = new HashMap<>();headers.put("type", "order");headers.put("priority", 1);headers.put("format", "json");// 设置消息属性（包含Headers）AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().headers(headers).build();// 发送消息（路由键被忽略，可填任意值）String message = "Order message with headers!";channel.basicPublish(EXCHANGE_NAME, "", // 路由键无意义props, message.getBytes("UTF-8"));System.out.println(" [x] Sent: '" + message + "'");}}
}

3、消费者（接收消息）

import com.rabbitmq.client.*;
import java.util.HashMap;
import java.util.Map;public class HeadersExchangeConsumer {private static final String EXCHANGE_NAME = "headers_logs";private static final String QUEUE_NAME = "order_high_priority";private static final String HOST = "10.106.182.54";private static final String USERNAME = "root";private static final String PASSWORD = "123456";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost(HOST);factory.setUsername(USERNAME);factory.setPassword(PASSWORD);Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明头交换机channel.exchangeDeclare(EXCHANGE_NAME, "headers");// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 定义绑定条件（需同时满足 type=order 且 priority=1）Map<String, Object> bindingArgs = new HashMap<>();bindingArgs.put("type", "order");bindingArgs.put("priority", 1);bindingArgs.put("x-match", "all"); // 匹配模式：all 或 any// 绑定队列到交换机channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", bindingArgs);System.out.println(" [*] Waiting for messages...");// 定义消息处理回调DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");Map<String, Object> headers = delivery.getProperties().getHeaders();System.out.println(" [x] Received: '" + message + "'");System.out.println("     Headers: " + headers);};// 监听队列消费消息channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}
}

队列绑定条件：

• 队列A绑定条件：type=order 且 priority=1（x-match=all）。
• 队列B绑定条件：format=json 或 format=xml（x-match=any）。

消息头匹配：

• 消息头 {type=order, priority=1, format=json} → 同时匹配队列A和队列B（若队列B的绑定条件为 x-match=any）。
• 消息头 {type=payment, format=json} → 仅匹配队列B（若绑定条件为 x-match=any）。

---

4、注意事项

• 键值对类型限制：消息头的值可以是字符串或数值类型，其他类型（如对象）需序列化处理。
• 头交换机的匹配逻辑较复杂（需遍历所有绑定条件），性能略低于直连或扇出交换机，但在合理设计下仍可高效运行。
• 避免过度复杂的绑定条件（如大量 x-match=all 的组合），可能增加匹配开销。
• 绑定队列时 routingKey 参数被忽略，但 RabbitMQ API 要求该参数存在（示例中传递空字符串）。

---

5、扩展应用

• 动态路由配置：根据消息头中的 version 和 region 动态路由到不同服务版本或地域的队列。 例如：version=2 且 region=us-east → 路由到 2版的 us-east。
• 灰度发布：通过消息头标记用户ID的哈希值（如 user_hash=30），将部分用户请求路由到新功能队列。
• 协议适配：消息头携带 content-type=avro，路由到 Avro 格式解析服务；携带 content-type=protobuf，路由到 Protobuf 解析服务。

头交换机通过灵活的消息头键值对匹配机制，为需要复杂多条件路由的场景提供了强大的支持。

---

五、四种交换机对比

交换机类型	路由规则	适用场景
Direct	精确匹配路由键	明确指定目标队列的场景
Fanout	广播到所有绑定队列（忽略路由键）	发布/订阅模式（如通知广播）
Topic	通配符匹配（* 或 #）	灵活的多条件路由（如分类消息）
Headers	基于消息头键值对匹配	复杂路由逻辑（不依赖路由键）