作者没有四次元口袋的蓝胖日期2026-10-17标签RabbitMQ, 发布订阅, Fanout, Direct, TopicRabbitMQ知识梳理(2)写在前面学完了 Hello World 和工作模型你会发现它们都有一个共同点——消息只发给一个消费者。但实际业务中我们经常需要一条消息发给多个系统或按条件筛选消息。这就是 RabbitMQ 的三大消息分发模式Fanout发布订阅广播给所有绑定的队列Direct路由精确匹配 routing keyTopic主题通配符匹配*和#这篇博客带你彻底搞懂这三种模式的区别和适用场景让你在面试时能从业务需求出发选择合适的消息分发方式。一、核心概念Exchange1.1 为什么需要Exchange在 Hello World 模型中生产者直接把消息发到队列。但在更复杂的场景中我们需要根据规则把消息分发到不同的队列。Exchange 的作用接收生产者的消息根据绑定规则Binding转发到对应的队列。生产者 → Exchange → (Binding规则) → Queue → 消费者1.2 Exchange的四种类型类型规则适用场景Fanout广播到所有绑定的队列消息广播如通知所有系统Directrouting key 精确匹配点对点路由如订单系统只发给库存系统Topicrouting key 通配符匹配多级路由如order.create发给订单系统Headers根据消息头属性匹配很少用性能差1.3 Binding绑定BindingExchange 和 Queue 之间的关联关系包含 routing key。// 绑定示例rabbitTemplate.convertAndSend(order.exchange,order.create,orderId);order.exchangeExchange 名称order.createrouting key绑定规则决定消息发到哪个队列二、Fanout模式发布订阅2.1 模型说明Fanout扇出把消息广播到所有绑定的队列忽略 routing key。┌→ Queue A → 消费者A订单系统 生产者 → Exchange ─┼→ Queue B → 消费者B库存系统 └→ Queue C → 消费者C积分系统适用场景消息需要发给多个系统如订单创建后库存、积分、物流系统都要知道。2.2 代码实现配置ConfigurationpublicclassFanoutConfig{// 创建 Fanout ExchangeBeanpublicFanoutExchangefanoutExchange(){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);}// 绑定队列到 ExchangeFanout 忽略 routing keyBeanpublicBindingbindingOrder(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 Exchangerouting key 被忽略rabbitTemplate.convertAndSend(order.fanout.exchange,,orderId);System.out.println(订单创建: orderId);}}消费者ComponentpublicclassOrderConsumers{// 订单系统消费者RabbitListener(queuesorder.fanout.queue.a)publicvoidorderSystem(LongorderId){System.out.println(订单系统 收到订单: orderId);// 记录订单日志}// 库存系统消费者RabbitListener(queuesorder.fanout.queue.b)publicvoidinventorySystem(LongorderId){System.out.println(库存系统 收到订单: orderId);// 扣减库存}// 积分系统消费者RabbitListener(queuesorder.fanout.queue.c)publicvoidpointSystem(LongorderId){System.out.println(积分系统 收到订单: orderId);// 增加积分}}测试TestpublicvoidtestFanout()throwsInterruptedException{orderProducer.createOrder(1001L);Thread.sleep(1000);}输出订单创建: 1001 订单系统 收到订单: 1001 库存系统 收到订单: 1001 积分系统 收到订单: 1001特点一条消息三个系统都收到routing key 被忽略可以传空字符串新增系统只需要新建一个队列并绑定到 Exchange2.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 ExchangeBeanpublicDirectExchangedirectExchange(){returnnewDirectExchange(order.direct.exchange);}// 创建队列BeanpublicQueuecreateOrderQueue(){returnnewQueue(order.direct.queue.create);}BeanpublicQueuepayOrderQueue(){returnnewQueue(order.direct.queue.pay);}BeanpublicQueuecancelOrderQueue(){returnnewQueue(order.direct.queue.cancel);}// 绑定队列到 Exchange指定 routing keyBeanpublicBindingbindingCreate(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.createrabbitTemplate.convertAndSend(order.direct.exchange,order.create,orderId);System.out.println(订单创建: orderId);}publicvoidpayOrder(LongorderId){// routing key order.payrabbitTemplate.convertAndSend(order.direct.exchange,order.pay,orderId);System.out.println(订单支付: orderId);}publicvoidcancelOrder(LongorderId){// routing key order.cancelrabbitTemplate.convertAndSend(order.direct.exchange,order.cancel,orderId);System.out.println(订单取消: orderId);}}消费者ComponentpublicclassOrderDirectConsumers{RabbitListener(queuesorder.direct.queue.create)publicvoidhandleCreate(LongorderId){System.out.println(订单创建系统 收到订单: orderId);}RabbitListener(queuesorder.direct.queue.pay)publicvoidhandlePay(LongorderId){System.out.println(订单支付系统 收到订单: orderId);}RabbitListener(queuesorder.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退化为 Fanout3.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 ExchangeBeanpublicTopicExchangetopicExchange(){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(queueslog.topic.queue.order.one)publicvoidorderOneLevel(Stringmessage){System.out.println(一级订单日志: message);}RabbitListener(queueslog.topic.queue.order.all)publicvoidorderAllLevel(Stringmessage){System.out.println(所有订单日志: message);}RabbitListener(queueslog.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 通配符规则总结Binding Key匹配的 Routing Key不匹配的 Routing Keyorder.*order.create,order.payorder.create.fast两级,user.loginorder.#order.create,order.create.fast,order.a.b.cuser.login#.errororder.error,pay.error,a.b.errororder.create,error.order*.*order.create,user.loginorder.create.fast三级4.4 实际场景日志收集应用系统 → Topic Exchange ├→ routing key: order.* → 订单日志系统一级 ├→ routing key: order.# → 订单归档系统多级 ├→ routing key: #.error → 错误告警系统 └→ routing key: # → 全量日志系统优势灵活的消息路由不同系统按需订阅日志五、三种模式对比特性FanoutDirectTopic规则广播忽略 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七、写在最后学习建议先理解区别再写代码Fanout、Direct、Topic 的区别是面试高频题先搞清楚它们的应用场景再动手写代码。Topic 是难点也是亮点通配符*和#的区别很多人搞不清面试时能说清楚order.* 匹配一级order.# 匹配多级会比只说通配符更加分。实际项目中组合使用一个系统可能同时用到三种模式Fanout 做订单广播Direct 做任务路由Topic 做日志收集。高频面试题总结问题核心答案Exchange有什么作用接收生产者消息根据Binding规则转发到队列Fanout和Direct有什么区别Fanout广播到所有队列忽略routing keyDirect精确匹配routing keyTopic的通配符*和#有什么区别匹配一个单词order.匹配 order.create#匹配零个或多个单词order.# 匹配 order.create.fast什么场景用Fanout消息需要发给所有系统如订单广播什么场景用Direct消息只发给一个特定的系统如订单创建只发给库存系统什么场景用Topic消息需要按类别筛选如日志收集、只接收错误日志