本文深入探讨 RabbitMQ 中 Exchange、Queue、Routing Key 的协作机制，以及不同场景下的消息消费策略。
一、核心概念回顾
RabbitMQ 消息流转的核心链路：
1.1 Exchange 类型
| 类型 | 特点 | 使用场景 |
|---|---|---|
| direct | 精确匹配 routing key | 点对点消息，精确路由 |
| topic | 通配符匹配 routing key（*和#） | 灵活路由，多级分类 |
| fanout | 忽略 routing key，广播到所有绑定队列 | 广播通知，事件发布 |
| headers | 基于消息头属性匹配 | 复杂路由条件 |
1.2 关键关系图
二、一个 Exchange 对应多个 Routing Key
2.1 设计模式
一个 Exchange 可以通过不同的 Routing Key 路由到不同的队列，这是推荐的最佳实践。
mq: order: delete: exchange: order.delete.exchange routingKey: deleteAll: order.routing.delete.all # 删除全部 deletePart: order.routing.delete.partial # 部分删除2.2 架构示意
2.3 优势
逻辑聚合：同一业务域的消息统一管理
灵活路由：消费者按需订阅
易于扩展：新增类型只需添加 routing key
资源节约：减少 Exchange 数量
三、消息竞争 vs 消息广播
这是理解 RabbitMQ 消费逻辑的核心问题。
3.1 场景一：消息竞争（Work Queue 模式）
多个消费者绑定同一个 Queue→ 消息只会被其中一个消费者处理
⚠️ 消息只会被 A 或 B 其中一个消费（轮询分发）
适用场景：
任务分发、负载均衡
耗时任务的并行处理
同一服务的多实例部署
代码示例：
// 多个消费者订阅同一队列 - 竞争消费 @Service public class OrderProcessorA { @RabbitListener(queues = "order.process.queue") public void handleOrder(String message) { // 处理订单 - 与其他消费者竞争 log.info("实例A处理消息: {}", message); } } @Service public class OrderProcessorB { @RabbitListener(queues = "order.process.queue") // 同一队列 public void handleOrder(String message) { // 处理订单 - 与其他消费者竞争 log.info("实例B处理消息: {}", message); } }3.2 场景二：消息广播（Pub/Sub 模式）
每个消费者绑定独立的 Queue→ 所有消费者都能收到消息
✅ 两个服务都能收到完整的消息副本！
适用场景：
事件通知（用户登录、订单创建）
数据同步（多系统数据一致性）
日志收集（多个系统记录同一事件）
四、实现广播的三种方式
4.1 方式一：Topic Exchange + 独立队列
每个服务创建自己的队列，使用相同的 routing key 绑定。
@Configuration public class MqConfig { @Value("${mq.user.login.exchange}") private String loginExchange; @PostConstruct public void initMq() { Channel channel = connection.createChannel(); // 声明 topic 类型 exchange channel.exchangeDeclare(loginExchange, "topic", true); // 服务A 绑定自己的队列 channel.queueDeclare("service-a.login.queue", true, false, false, null); channel.queueBind("service-a.login.queue", loginExchange, "user.login.event"); // 服务B 绑定自己的队列（相同 routing key） channel.queueDeclare("service-b.login.queue", true, false, false, null); channel.queueBind("service-b.login.queue", loginExchange, "user.login.event"); } }4.2 方式二：Fanout Exchange（推荐用于纯广播）
Fanout 类型忽略 routing key，消息直接广播到所有绑定的队列。
@PostConstruct public void initBroadcastMq() { Channel channel = connection.createChannel(); // 声明 fanout 类型 - 广播模式 channel.exchangeDeclare("user.logout.fanout", "fanout", true); // 各服务绑定自己的队列（routing key 为空） channel.queueDeclare("order-service.logout.queue", true, false, false, null); channel.queueBind("order-service.logout.queue", "user.logout.fanout", ""); channel.queueDeclare("cart-service.logout.queue", true, false, false, null); channel.queueBind("cart-service.logout.queue", "user.logout.fanout", ""); channel.queueDeclare("session-service.logout.queue", true, false, false, null); channel.queueBind("session-service.logout.queue", "user.logout.fanout", ""); }4.3 方式三：临时队列（适合临时消费者）
使用自动生成的队列名，服务停止后队列自动删除。
// 自动生成唯一队列名 String queueName = channel.queueDeclare().getQueue(); // 如: amq.gen-JzTY20BRgKO-HjmUJj0wLg // 绑定到 exchange channel.queueBind(queueName, "event.exchange", "order.created"); // 服务停止后，队列自动删除五、配置示例与最佳实践
5.1 完整配置示例
mq: # 基础配置 host: 127.0.0.1 username: admin password: admin123 # 业务 Exchange 配置 order: # 订单处理 - 竞争消费模式 process: exchange: order.process.exchange queue: order.process.queue # 多实例共享队列 routingKey: order.routing.process # 订单删除 - 多类型路由 delete: exchange: order.delete.exchange routingKey: deleteAll: order.routing.delete.all deletePart: order.routing.delete.partial # 用户事件 - 广播模式 user: logout: exchange: user.logout.fanout # fanout 类型 queue: ${spring.application.name}.logout.queue # 每个服务独立队列 login: exchange: user.login.fanout queue: ${spring.application.name}.login.queue5.2 Exchange 初始化代码
@Service public class RabbitMqInitializer { private static final Logger log = LoggerFactory.getLogger(RabbitMqInitializer.class); @Resource private ConnectionFactory connectionFactory; @Value("${mq.order.process.exchange}") private String processExchange; @Value("${mq.order.process.queue}") private String processQueue; @Value("${mq.order.process.routingKey}") private String processRoutingKey; @Value("${mq.user.logout.exchange}") private String logoutExchange; @Value("${mq.user.logout.queue}") private String logoutQueue; @PostConstruct public void initializeQueues() { try { Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); // ========== 订单处理 - Topic Exchange ========== channel.exchangeDeclare(processExchange, "topic", true, false, null); channel.queueDeclare(processQueue, true, false, false, null); channel.queueBind(processQueue, processExchange, processRoutingKey); log.info("订单处理MQ初始化完成 - Exchange: {}, Queue: {}", processExchange, processQueue); // ========== 用户登出 - Fanout Exchange (广播) ========== channel.exchangeDeclare(logoutExchange, "fanout", true, false, null); channel.queueDeclare(logoutQueue, true, false, false, null); channel.queueBind(logoutQueue, logoutExchange, ""); // fanout 不需要 routing key log.info("用户登出MQ初始化完成 - Exchange: {}, Queue: {}", logoutExchange, logoutQueue); channel.close(); } catch (Exception e) { log.error("MQ初始化失败", e); } } }5.3 消息生产者
@Service public class MessageProducer { private static final Logger log = LoggerFactory.getLogger(MessageProducer.class); @Resource private RabbitMqInitializer rabbitMqInitializer; @Value("${mq.order.delete.exchange}") private String deleteExchange; @Value("${mq.order.delete.routingKey.deleteAll}") private String deleteAllRoutingKey; @Value("${mq.order.delete.routingKey.deletePart}") private String deletePartRoutingKey; /** * 发送删除全部订单消息 */ public void publishDeleteAll(String message) { log.info("发送删除全部消息: {}", message); publishMessage(message, deleteExchange, deleteAllRoutingKey); } /** * 发送部分删除消息 */ public void publishDeletePart(String message) { log.info("发送部分删除消息: {}", message); publishMessage(message, deleteExchange, deletePartRoutingKey); } private void publishMessage(String message, String exchange, String routingKey) { try { Channel channel = getChannel(); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .deliveryMode(2) // 持久化 .build(); channel.basicPublish(exchange, routingKey, props, message.getBytes()); log.info("消息发送成功 - Exchange: {}, RoutingKey: {}", exchange, routingKey); } catch (Exception e) { log.error("消息发送失败: {}", message, e); } } }六、常见问题与解决方案
Q1: 如何确保消息不丢失？
// 1. Exchange 持久化 channel.exchangeDeclare(exchange, "topic", true); // durable=true // 2. Queue 持久化 channel.queueDeclare(queue, true, false, false, null); // durable=true // 3. 消息持久化 AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .deliveryMode(2) // 2=持久化 .build(); // 4. 开启发布确认 channel.confirmSelect(); channel.basicPublish(exchange, routingKey, props, message.getBytes()); if (!channel.waitForConfirms()) { log.error("消息发布确认失败"); }Q2: 如何控制消费速率？
// 设置 prefetch count，限制未确认消息数量 channel.basicQos(1); // 每次只预取1条消息 // 手动确认 channel.basicConsume(queue, false, (consumerTag, delivery) -> { try { // 处理消息 processMessage(delivery.getBody()); // 手动确认 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } catch (Exception e) { // 拒绝并重新入队 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true); } }, consumerTag -> {});Q3: 同一服务多实例如何避免重复消费？
答：让多实例共享同一个 Queue 名称即可，RabbitMQ 会自动轮询分发。
# 所有实例使用相同的队列名 mq: order: process: queue: order.process.queue # 固定队列名，不要用 ${instance.id}七、完整流程图
7.1 消息发布流程
7.2 竞争消费 vs 广播消费对比
八、总结
| 消费模式 | Queue 策略 | Exchange 类型 | 适用场景 |
|---|---|---|---|
| 竞争消费 | 多消费者共享同一 Queue | 任意 | 任务分发、负载均衡 |
| 广播消费 | 每个消费者独立 Queue | fanout / topic | 事件通知、数据同步 |
| 选择性消费 | 独立 Queue + 不同 routing key | topic / direct | 按类型订阅 |
核心原则：
消息是发送到Queue的，不是直接发送到消费者的。 同一个 Queue 的消息只会被消费一次，不同 Queue 可以收到相同消息的副本。
理解这一点，就能灵活设计各种消息消费场景！
