企业官网建设流程全解析

作者：没有四次元口袋的蓝胖
日期：2026-10-17
标签：RabbitMQ, 发布订阅, Fanout, Direct, Topic

RabbitMQ知识梳理(2)

写在前面

学完了 Hello World 和工作模型，你会发现它们都有一个共同点——消息只发给一个消费者。但实际业务中，我们经常需要"一条消息发给多个系统"或"按条件筛选消息"。

这就是 RabbitMQ 的三大消息分发模式：

1. Fanout（发布订阅）：广播给所有绑定的队列
2. Direct（路由）：精确匹配 routing key
3. Topic（主题）：通配符匹配（*和#）

这篇博客带你彻底搞懂这三种模式的区别和适用场景，让你在面试时能从业务需求出发，选择合适的消息分发方式。

一、核心概念：Exchange

1.1 为什么需要Exchange？

在 Hello World 模型中，生产者直接把消息发到队列。但在更复杂的场景中，我们需要根据规则把消息分发到不同的队列。

Exchange 的作用：接收生产者的消息，根据绑定规则（Binding）转发到对应的队列。

生产者 → Exchange → (Binding规则) → Queue → 消费者

1.2 Exchange的四种类型

1.3 Binding（绑定）

Binding：Exchange 和 Queue 之间的关联关系，包含 routing key。

// 绑定示例rabbitTemplate.convertAndSend("order.exchange","order.create",orderId);

• order.exchange：Exchange 名称
• order.create：routing key
• 绑定规则决定消息发到哪个队列

二、Fanout模式：发布订阅

2.1 模型说明

Fanout（扇出）：把消息广播到所有绑定的队列，忽略 routing key。

┌→ Queue A → 消费者A（订单系统） 生产者 → Exchange ─┼→ Queue B → 消费者B（库存系统） └→ Queue C → 消费者C（积分系统）

适用场景：消息需要发给多个系统（如订单创建后，库存、积分、物流系统都要知道）。

2.2 代码实现

配置

@ConfigurationpublicclassFanoutConfig{// 创建 Fanout Exchange@BeanpublicFanoutExchangefanoutExchange(){returnnewFanoutExchange("order.fanout.exchange");}// 创建队列A（订单系统）@BeanpublicQueueorderQueue(){returnnewQueue("order.fanout.queue.a");}// 创建队列B（库存系统）@BeanpublicQueueinventoryQueue(){returnnewQueue("order.fanout.queue.b");}// 创建队列C（积分系统）@BeanpublicQueuepointQueue(){returnnewQueue("order.fanout.queue.c");}// 绑定队列到 Exchange（Fanout 忽略 routing key）@BeanpublicBindingbindingOrder(QueueorderQueue,FanoutExchangefanoutExchange){returnBindingBuilder.bind(orderQueue).to(fanoutExchange);}@BeanpublicBindingbindingInventory(QueueinventoryQueue,FanoutExchangefanoutExchange){returnBindingBuilder.bind(inventoryQueue).to(fanoutExchange);}@BeanpublicBindingbindingPoint(QueuepointQueue,FanoutExchangefanoutExchange){returnBindingBuilder.bind(pointQueue).to(fanoutExchange);}}

生产者

@ComponentpublicclassOrderProducer{@AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidcreateOrder(LongorderId){// 发送消息到 Fanout Exchange（routing key 被忽略）rabbitTemplate.convertAndSend("order.fanout.exchange","",orderId);System.out.println("订单创建: "+orderId);}}

消费者

@ComponentpublicclassOrderConsumers{// 订单系统消费者@RabbitListener(queues="order.fanout.queue.a")publicvoidorderSystem(LongorderId){System.out.println("订单系统 收到订单: "+orderId);// 记录订单日志}// 库存系统消费者@RabbitListener(queues="order.fanout.queue.b")publicvoidinventorySystem(LongorderId){System.out.println("库存系统 收到订单: "+orderId);// 扣减库存}// 积分系统消费者@RabbitListener(queues="order.fanout.queue.c")publicvoidpointSystem(LongorderId){System.out.println("积分系统 收到订单: "+orderId);// 增加积分}}

测试

@TestpublicvoidtestFanout()throwsInterruptedException{orderProducer.createOrder(1001L);Thread.sleep(1000);}

输出：

订单创建: 1001 订单系统 收到订单: 1001 库存系统 收到订单: 1001 积分系统 收到订单: 1001

特点：

• 一条消息，三个系统都收到
• routing key 被忽略（可以传空字符串）
• 新增系统只需要新建一个队列并绑定到 Exchange

2.3 实际场景：订单广播

用户下单 → 订单系统发送消息到 Fanout Exchange ├→ 库存系统：扣减库存 ├→ 积分系统：增加积分 ├→ 物流系统：准备发货 └→ 风控系统：风险评估

优势：

• 订单系统不需要知道有哪些下游系统
• 新增系统（如风控）只需要新建一个消费者，订单系统不用改

三、Direct模式：路由

3.1 模型说明

Direct（直连）：routing key精确匹配，消息只发给匹配的队列。

┌→ Queue A (routing key: order.create) → 订单系统 生产者 → Exchange ──────┼→ Queue B (routing key: order.pay) → 支付系统 └→ Queue C (routing key: order.cancel) → 取消系统

适用场景：消息只需要发给一个特定的系统（如订单创建只发给库存系统）。

3.2 代码实现

配置

@ConfigurationpublicclassDirectConfig{// 创建 Direct Exchange@BeanpublicDirectExchangedirectExchange(){returnnewDirectExchange("order.direct.exchange");}// 创建队列@BeanpublicQueuecreateOrderQueue(){returnnewQueue("order.direct.queue.create");}@BeanpublicQueuepayOrderQueue(){returnnewQueue("order.direct.queue.pay");}@BeanpublicQueuecancelOrderQueue(){returnnewQueue("order.direct.queue.cancel");}// 绑定队列到 Exchange（指定 routing key）@BeanpublicBindingbindingCreate(QueuecreateOrderQueue,DirectExchangedirectExchange){returnBindingBuilder.bind(createOrderQueue).to(directExchange).with("order.create");// routing key}@BeanpublicBindingbindingPay(QueuepayOrderQueue,DirectExchangedirectExchange){returnBindingBuilder.bind(payOrderQueue).to(directExchange).with("order.pay");}@BeanpublicBindingbindingCancel(QueuecancelOrderQueue,DirectExchangedirectExchange){returnBindingBuilder.bind(cancelOrderQueue).to(directExchange).with("order.cancel");}}

生产者

@ComponentpublicclassOrderRouter{@AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidcreateOrder(LongorderId){// routing key = "order.create"rabbitTemplate.convertAndSend("order.direct.exchange","order.create",orderId);System.out.println("订单创建: "+orderId);}publicvoidpayOrder(LongorderId){// routing key = "order.pay"rabbitTemplate.convertAndSend("order.direct.exchange","order.pay",orderId);System.out.println("订单支付: "+orderId);}publicvoidcancelOrder(LongorderId){// routing key = "order.cancel"rabbitTemplate.convertAndSend("order.direct.exchange","order.cancel",orderId);System.out.println("订单取消: "+orderId);}}

消费者

@ComponentpublicclassOrderDirectConsumers{@RabbitListener(queues="order.direct.queue.create")publicvoidhandleCreate(LongorderId){System.out.println("订单创建系统 收到订单: "+orderId);}@RabbitListener(queues="order.direct.queue.pay")publicvoidhandlePay(LongorderId){System.out.println("订单支付系统 收到订单: "+orderId);}@RabbitListener(queues="order.direct.queue.cancel")publicvoidhandleCancel(LongorderId){System.out.println("订单取消系统 收到订单: "+orderId);}}

测试

@TestpublicvoidtestDirect()throwsInterruptedException{orderRouter.createOrder(1001L);// 只发给创建系统orderRouter.payOrder(1001L);// 只发给支付系统orderRouter.cancelOrder(1002L);// 只发给取消系统Thread.sleep(1000);}

输出：

订单创建: 1001 订单创建系统 收到订单: 1001 订单支付: 1001 订单支付系统 收到订单: 1001 订单取消: 1002 订单取消系统 收到订单: 1002

特点：

• routing key 精确匹配
• 一个 routing key 只能绑定一个队列（一对一）
• 如果多个队列绑定同一个 routing key，退化为 Fanout

3.3 实际场景：订单路由

订单系统 → Direct Exchange ├→ routing key: order.create → 库存系统 ├→ routing key: order.pay → 支付系统 └→ routing key: order.cancel → 取消系统

优势：

• 精确控制消息路由
• 不同类型的订单事件发给不同的系统

四、Topic模式：主题

4.1 模型说明

Topic（主题）：routing key 使用通配符匹配，支持多级路由。

routing key 格式：单词1.单词2.单词3（用.分隔）

通配符：

• *：匹配一个单词
• #：匹配零个或多个单词

┌→ Queue A (binding: order.*) → 匹配 order.create, order.pay 生产者 → Exchange ──────┼→ Queue B (binding: order.#) → 匹配 order.create, order.create.fast └→ Queue C (binding: #.error) → 匹配 order.error, pay.error

适用场景：消息需要按类别筛选（如所有订单事件、所有错误日志）。

4.2 代码实现

配置

@ConfigurationpublicclassTopicConfig{// 创建 Topic Exchange@BeanpublicTopicExchangetopicExchange(){returnnewTopicExchange("log.topic.exchange");}// 队列A：只接收 order.* 的日志（一级）@BeanpublicQueueorderOneLevelQueue(){returnnewQueue("log.topic.queue.order.one");}// 队列B：接收 order.* 的所有日志（多级）@BeanpublicQueueorderAllLevelQueue(){returnnewQueue("log.topic.queue.order.all");}// 队列C：接收所有错误日志@BeanpublicQueueerrorQueue(){returnnewQueue("log.topic.queue.error");}// 绑定规则@BeanpublicBindingbindingOrderOne(QueueorderOneLevelQueue,TopicExchangetopicExchange){returnBindingBuilder.bind(orderOneLevelQueue).to(topicExchange).with("order.*");// 只匹配一级（如 order.create）}@BeanpublicBindingbindingOrderAll(QueueorderAllLevelQueue,TopicExchangetopicExchange){returnBindingBuilder.bind(orderAllLevelQueue).to(topicExchange).with("order.#");// 匹配多级（如 order.create, order.create.fast）}@BeanpublicBindingbindingError(QueueerrorQueue,TopicExchangetopicExchange){returnBindingBuilder.bind(errorQueue).to(topicExchange).with("#.error");// 匹配所有以 .error 结尾的}}

生产者

@ComponentpublicclassLogProducer{@AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidsendLog(StringroutingKey,Stringmessage){rabbitTemplate.convertAndSend("log.topic.exchange",routingKey,message);System.out.println("发送日志 ["+routingKey+"]: "+message);}}

消费者

@ComponentpublicclassLogConsumers{@RabbitListener(queues="log.topic.queue.order.one")publicvoidorderOneLevel(Stringmessage){System.out.println("一级订单日志: "+message);}@RabbitListener(queues="log.topic.queue.order.all")publicvoidorderAllLevel(Stringmessage){System.out.println("所有订单日志: "+message);}@RabbitListener(queues="log.topic.queue.error")publicvoiderrorLog(Stringmessage){System.out.println("错误日志: "+message);}}

测试

@TestpublicvoidtestTopic()throwsInterruptedException{// 一级订单日志logProducer.sendLog("order.create","订单创建");logProducer.sendLog("order.pay","订单支付");// 多级订单日志logProducer.sendLog("order.create.fast","快速订单创建");// 错误日志logProducer.sendLog("order.error","订单处理失败");logProducer.sendLog("pay.error","支付失败");// 其他日志logProducer.sendLog("user.login","用户登录");Thread.sleep(1000);}

输出：

发送日志 [order.create]: 订单创建 一级订单日志: 订单创建 ← order.* 匹配 所有订单日志: 订单创建 ← order.# 匹配 发送日志 [order.pay]: 订单支付 一级订单日志: 订单支付 ← order.* 匹配 所有订单日志: 订单支付 ← order.# 匹配 发送日志 [order.create.fast]: 快速订单创建 所有订单日志: 快速订单创建 ← order.# 匹配（order.* 不匹配，因为是两级） 发送日志 [order.error]: 订单处理失败 一级订单日志: 订单处理失败 ← order.* 匹配 所有订单日志: 订单处理失败 ← order.# 匹配 错误日志: 订单处理失败 ← #.error 匹配 发送日志 [pay.error]: 支付失败 错误日志: 支付失败 ← #.error 匹配 发送日志 [user.login]: 用户登录 （没有消费者接收，消息被丢弃）

4.3 通配符规则总结

4.4 实际场景：日志收集

应用系统 → Topic Exchange ├→ routing key: order.* → 订单日志系统（一级） ├→ routing key: order.# → 订单归档系统（多级） ├→ routing key: #.error → 错误告警系统 └→ routing key: # → 全量日志系统

优势：

• 灵活的消息路由
• 不同系统按需订阅日志

五、三种模式对比

选择建议：

• Fanout：消息需要发给所有系统（如订单创建后，所有下游系统都要知道）
• Direct：消息只发给一个特定的系统（如订单创建只发给库存系统）
• Topic：消息需要按类别筛选（如只接收订单相关的日志，或只接收错误日志）

六、思维导图速览

RabbitMQ三大消息分发模式 ├── Exchange（消息交换机） │ ├── 接收生产者消息 │ ├── 根据 Binding 规则转发到队列 │ └── 四种类型：Fanout / Direct / Topic / Headers ├── Fanout（发布订阅） │ ├── 规则：广播到所有绑定的队列，忽略 routing key │ ├── 适用：消息广播（如订单通知所有系统） │ └── 代码：BindingBuilder.bind(queue).to(fanoutExchange) ├── Direct（路由） │ ├── 规则：routing key 精确匹配 │ ├── 适用：点对点路由（如订单创建只发给库存系统） │ └── 代码：BindingBuilder.bind(queue).to(directExchange).with("order.create") ├── Topic（主题） │ ├── 规则：routing key 通配符匹配 │ │ ├── *：匹配一个单词 │ │ └── #：匹配零个或多个单词 │ ├── 适用：多级路由、分类订阅（如日志收集） │ └── 代码：BindingBuilder.bind(queue).to(topicExchange).with("order.*") └── 选择建议 ├── 广播所有系统 → Fanout ├── 精确路由到一个系统 → Direct └── 按类别筛选 → Topic

七、写在最后

学习建议

1. 先理解区别，再写代码
   Fanout、Direct、Topic 的区别是面试高频题，先搞清楚它们的应用场景，再动手写代码。

2. Topic 是难点也是亮点
   通配符*和#的区别很多人搞不清，面试时能说清楚"order.* 匹配一级，order.# 匹配多级"，会比只说"通配符"更加分。

3. 实际项目中组合使用
   一个系统可能同时用到三种模式：Fanout 做订单广播，Direct 做任务路由，Topic 做日志收集。

高频面试题总结