1. 从一条消息说起为什么需要“一发多收”做过消息队列开发的人大概率都遇到过这样的场景用户下单成功后系统需要同时做几件事——给用户发短信、给仓库推发货单、给数据分析平台记一笔流水。最直觉的做法是在下单逻辑里依次调用这三个服务但这样一来下单接口的响应时间就等于三个服务的耗时之和任何一个环节卡住整个下单流程都得跟着等。更麻烦的是如果短信服务临时挂了你是让下单失败还是吞掉异常吞掉异常用户收不到短信不吞异常用户下不了单两头为难。消息队列就是来解决这类问题的。生产者只管把“订单已创建”这条消息丢出去至于谁来消费、消费几条、消费失败怎么办统统不用操心。而RabbitMQ 的 Topic 交换机正是实现“一条消息发往多个队列”的核心武器。它的工作模型不复杂生产者把消息发给交换机交换机根据路由键和绑定关系的匹配规则把同一条消息复制分发到多个队列里每个队列背后可以挂不同的消费者各干各的活互不干扰。这篇内容要聊的就是怎么在 SpringBoot 项目里把这套机制落地。具体来说我会带你走一遍完整流程从交换机、队列、绑定的声明到用注解方式写多个接收器再到路由键的设计技巧和实际踩过的坑。适合已经了解 RabbitMQ 基本概念、想在 SpringBoot 里把“一发多收”真正用起来的开发者。如果你还没接触过 RabbitMQ建议先补一下生产者、消费者、交换机这几个基础概念不然直接看绑定规则会有点懵。需要提前说明的是标题里提到的“接收器”在 SpringBoot 里通常有两种写法一种是RabbitListener注解直接标在方法上另一种是手动注册MessageListener容器。前者写起来清爽后者灵活度更高。我会以注解方式为主线因为它更符合 SpringBoot 的开发习惯同时在关键位置补充手动配置的替代方案方便你根据项目情况取舍。2. 核心机制拆解Topic 交换机到底怎么“一对多”2.1 交换机、队列、绑定三者的关系要理解“一条消息发往多个队列”得先把 RabbitMQ 里三个核心角色的分工理清楚。很多人初学时容易把交换机和队列混为一谈其实它们职责完全不同。交换机Exchange是消息进入 RabbitMQ 后的第一站它本身不存储消息只负责根据规则做转发。你可以把它想象成邮局的分拣中心信件到了分拣中心工作人员看地址决定投到哪个片区的邮筒。队列Queue才是真正存消息的地方消息在队列里排队等待消费者取走。绑定Binding则是交换机和队列之间的“路由规则”它告诉交换机什么样的消息该送到哪个队列。Topic 交换机的绑定规则有个特点路由键支持通配符匹配。这里有两个符号必须记牢*匹配一个单词以点分隔的一段#匹配零个或多个单词举个例子路由键是order.create.success那么绑定键order.*.success能匹配上*吃掉中间的createorder.#也能匹配上#吃掉后面所有但order.*匹配不上因为*只能吃一个单词而后面还剩create.success两段。正是这个通配符机制让同一条消息可以被多个不同绑定键的队列同时命中。比如一条路由键为order.create.success的消息队列 A 绑定了order.#队列 B 绑定了order.*.success队列 C 绑定了#.success那么这条消息会被复制三份分别进入 A、B、C 三个队列。这就是“一发多收”的底层原理。2.2 为什么选 Topic 而不是 Direct 或 FanoutRabbitMQ 有四种交换机类型选哪种直接决定了你的路由能力上限。我简单对比一下你就明白为什么这个场景非 Topic 莫属。交换机类型路由规则适用场景一对多能力Direct路由键完全相等才匹配点对点精确投递支持但需要多个绑定键完全相同Fanout忽略路由键广播到所有绑定队列全量广播支持但无法做筛选Topic通配符模糊匹配按业务维度灵活分发支持且可精细控制Headers按消息头匹配复杂条件匹配支持但性能较差、用得少Direct 交换机其实也能实现一对多只要把多个队列绑定到同一个路由键就行。但它的短板在于“不够灵活”——一旦业务需要按不同维度筛选你就得为每种组合单独建绑定维护成本飙升。Fanout 倒是简单粗暴所有队列都能收到可它完全没法筛选短信队列和日志队列收到的是同样的消息这显然不合理。Topic 的好处在于它用一套通配符规则覆盖了从精确到模糊的各种匹配需求。你可以用order.create精确投递也可以用order.#接收所有订单相关消息还能用#.error接收所有错误日志。一套机制多种玩法这就是它在实际项目里出场率最高的原因。2.3 消息复制的代价一对多不是没有成本这里必须泼一盆冷水。很多人以为“一条消息发往多个队列”是零成本的实际上 RabbitMQ 在背后做了消息复制。交换机每命中一个队列就会把消息内容写一份到那个队列里。三个队列就是三份存储十个队列就是十份。这意味着两件事。第一磁盘和内存开销随队列数量线性增长。如果你的消息体很大比如带附件、带大 JSON又绑定了很多队列broker 的压力会明显上升。第二消费速度受最慢的队列拖累。虽然各队列独立消费但如果某个队列积压严重它会持续占用 broker 资源间接影响其他队列的表现。所以我的经验是绑定到同一个交换机的队列数量要克制通常控制在个位数。如果发现某个业务需要广播给几十个消费者那可能不是消息队列该干的事考虑换成事件总线或者专门的推送系统更合适。3. SpringBoot 环境搭建与基础配置3.1 依赖引入与连接参数SpringBoot 集成 RabbitMQ 靠的是spring-boot-starter-amqp这个 starter它把 RabbitMQ 的 Java 客户端和 Spring 的抽象层都打包好了。在pom.xml里加这么一段dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency版本不用自己指定跟着 SpringBoot 父工程的版本管理走就行。如果你用的是 Gradle对应写法是implementation org.springframework.boot:spring-boot-starter-amqp。接下来是连接配置写在application.yml里spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest virtual-host: / listener: simple: acknowledge-mode: manual prefetch: 1 concurrency: 2 max-concurrency: 5这里有几个参数值得展开说。acknowledge-mode设成manual表示手动确认消费者处理完业务逻辑后自己调basicAck这样能避免消息丢失。如果你图省事用auto消息一投递出去就被标记为已消费业务处理失败也不会重投风险很大。prefetch设成 1 表示每个消费者一次只预取一条消息处理完再取下一条这样能保证负载均衡不会出现某个消费者囤了一堆消息而其他消费者闲着的情况。concurrency和max-concurrency控制消费者线程数的下限和上限根据你的业务耗时和机器配置调整。提示guest账号默认只允许本机连接生产环境务必创建独立账号并设置合理权限同时开启虚拟主机隔离不同业务。3.2 声明交换机、队列与绑定SpringBoot 里声明这些组件有两种方式用Bean在配置类里声明或者用RabbitListener注解自动声明。我推荐前者因为集中管理更清晰排查问题时一眼能看到全貌。Configuration public class RabbitConfig { public static final String TOPIC_EXCHANGE order.topic.exchange; public static final String SMS_QUEUE order.sms.queue; public static final String WAREHOUSE_QUEUE order.warehouse.queue; public static final String ANALYTICS_QUEUE order.analytics.queue; Bean public TopicExchange orderExchange() { return new TopicExchange(TOPIC_EXCHANGE, true, false); } Bean public Queue smsQueue() { return new Queue(SMS_QUEUE, true); } Bean public Queue warehouseQueue() { return new Queue(WAREHOUSE_QUEUE, true); } Bean public Queue analyticsQueue() { return new Queue(ANALYTICS_QUEUE, true); } Bean public Binding smsBinding() { return BindingBuilder.bind(smsQueue()) .to(orderExchange()) .with(order.#); } Bean public Binding warehouseBinding() { return BindingBuilder.bind(warehouseQueue()) .to(orderExchange()) .with(order.create.*); } Bean public Binding analyticsBinding() { return BindingBuilder.bind(analyticsQueue()) .to(orderExchange()) .with(#.success); } }这段代码里TopicExchange的第二个参数true表示持久化broker 重启后交换机还在第三个参数false表示不自动删除。队列的true同样表示持久化。绑定关系用BindingBuilder链式构建.with()后面跟的就是绑定键。三个队列的绑定键设计是有讲究的。短信队列用order.#意思是所有订单相关消息它都要因为用户可能对下单、取消、退款都关心。仓库队列用order.create.*只关心创建类消息取消和退款跟它无关。分析队列用#.success只收成功类事件失败的不进分析库。这样一条order.create.success的消息三个队列全都能命中各取所需。3.3 消息发送端的写法发送端用RabbitTemplate就够了SpringBoot 会自动注入。核心代码如下Service public class OrderService { Autowired private RabbitTemplate rabbitTemplate; public void createOrder(Order order) { // 业务逻辑省略 String routingKey order.create.success; rabbitTemplate.convertAndSend( RabbitConfig.TOPIC_EXCHANGE, routingKey, order, message - { message.getMessageProperties() .setMessageId(UUID.randomUUID().toString()); message.getMessageProperties() .setContentType(application/json); return message; } ); } }convertAndSend的四个参数分别是交换机名、路由键、消息体、消息后置处理器。后置处理器里我设置了消息 ID 和内容类型消息 ID 用于幂等去重内容类型告诉消费者怎么反序列化。如果你发送的是对象SpringBoot 默认用 JDK 序列化但 JDK 序列化跨语言兼容性差建议换成 JacksonBean public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); }把这个 Bean 注册进去RabbitTemplate和监听器都会自动用它做序列化省心。4. 多个接收器的实现与路由键设计4.1 用 RabbitListener 写三个消费者接收器的核心就是RabbitListener注解标在方法上指定监听的队列名。三个队列对应三个方法各写各的业务逻辑Component public class OrderMessageReceiver { RabbitListener(queues RabbitConfig.SMS_QUEUE) public void handleSms(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(短信服务收到订单 order.getOrderNo()); // 调用短信发送逻辑 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); } } RabbitListener(queues RabbitConfig.WAREHOUSE_QUEUE) public void handleWarehouse(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(仓库服务收到订单 order.getOrderNo()); // 生成发货单 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, true); } } RabbitListener(queues RabbitConfig.ANALYTICS_QUEUE) public void handleAnalytics(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(分析服务收到订单 order.getOrderNo()); // 写入分析库 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); } } }每个方法都接收Channel和deliveryTag这是手动确认的前提。basicAck表示确认消费basicNack的第三个参数表示是否重新入队。注意仓库那个方法我传了true因为发货单生成失败通常是临时问题比如数据库连接抖动重试有意义而短信和分析失败传false避免无限重试把队列堵死。注意手动确认模式下如果消费者方法抛异常且你没捕获SpringBoot 默认会拒绝消息并重新入队可能导致死循环。所以务必用 try-catch 包住业务逻辑明确控制 ack 和 nack。4.2 路由键设计的三个实用原则路由键设计得好不好直接决定了后期扩展是轻松还是痛苦。我总结了三条原则都是从实际项目里踩出来的。第一用点分层级从粗到细。比如order.create.success第一段是业务域第二段是动作第三段是结果。这样绑定的时候可以按任意层级筛选。反过来写成createOrderSuccess就没法拆了只能全匹配。第二预留扩展位。我习惯在业务域后面留一段表示来源或渠道比如order.app.create.success、order.web.create.success。这样以后想单独处理 App 渠道的订单绑定order.app.#就行不用改生产者代码。第三避免用#开头做绑定。虽然#.success语法上合法但它会匹配所有以 success 结尾的消息包括其他业务域的。如果交换机是共享的很容易误收。建议绑定键至少带上业务域前缀比如order.#.success。4.3 一个消费者监听多个队列的写法有时候几个队列的处理逻辑高度相似没必要写多个方法。RabbitListener的queues属性支持数组可以一次监听多个RabbitListener(queues {RabbitConfig.SMS_QUEUE, RabbitConfig.ANALYTICS_QUEUE}) public void handleMultiple(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { // 根据消息来源做不同处理 channel.basicAck(tag, false); }但这里有个坑多个队列共用一个消费者方法时你没法直接知道消息是从哪个队列来的。解决办法是从消息头里取AmqpHeaders.CONSUMER_QUEUE或者干脆拆成两个方法。我的建议是除非逻辑真的完全一样否则还是分开写可读性更好排查问题也方便。5. 实操验证与常见问题排查5.1 完整验证流程配置写完后怎么确认“一发多收”真的生效了我通常按这个顺序验证。第一步启动项目去 RabbitMQ 管理后台默认 15672 端口看 Exchanges 和 Queues 页面确认交换机和三个队列都创建成功绑定关系也在 Bindings 里能看到。如果队列没出现多半是配置类没被扫描到检查一下包路径。第二步调一次下单接口观察控制台。三个接收器方法应该都打印出订单号说明消息被复制分发到了三个队列。如果只有部分打印去管理后台看对应队列的 Ready 数量有堆积说明消费者没起来或者绑定键写错了。第三步故意在某个接收器里抛异常看消息是否按预期 nack。比如让短信方法抛异常并basicNack(tag, false, false)然后去后台看这个队列的消息是不是被丢弃了或者进了死信队列如果你配了的话而其他两个队列不受影响。这一步能验证队列之间的隔离性。5.2 常见问题速查表现象可能原因排查方向消息只进了一个队列绑定键不匹配检查路由键和绑定键的通配符是否对得上三个队列都收不到消息交换机名或路由键写错对比发送端和配置类的常量消费者方法不执行队列没绑定或监听器没注册看后台队列的 Consumers 数量是否为 0消息重复消费ack 时机不对或网络抖动引入消息 ID 做幂等检查 ack 是否在业务后启动报连接拒绝broker 没启动或端口不对确认 5672 端口可达账号密码正确消息体反序列化失败序列化方式不一致统一用 Jackson2JsonMessageConverter5.3 几个容易忽略的细节关于消息顺序。一条消息复制到三个队列后三个队列的消费是并行的谁先处理完不确定。如果你有严格的顺序要求比如必须先记流水再发短信那不能靠队列来保证得在业务层加状态机或者用单队列串行消费。关于死信队列。生产环境强烈建议给每个业务队列配死信队列。消息被 nack 且不重新入队时会进死信队列方便事后排查。配置方式是在声明队列时加参数Bean public Queue smsQueue() { return QueueBuilder.durable(SMS_QUEUE) .withArgument(x-dead-letter-exchange, order.dlx.exchange) .withArgument(x-dead-letter-routing-key, order.sms.dead) .build(); }关于消费者线程数。concurrency不是越大越好。如果业务是 CPU 密集型线程数超过 CPU 核心数反而增加上下文切换开销如果是 IO 密集型可以适当调大。我一般从 2 开始压测后根据吞吐量调整。关于消息体大小。RabbitMQ 默认单条消息最大 128MB但实际别发这么大的。超过 1MB 的消息就会明显影响性能建议把大内容存到对象存储消息里只传引用地址。6. 我踩过的坑和几条实在建议说几个我在实际项目里真金白银换来的教训。第一个坑是绑定键写反了通配符位置。有次我把order.create.*写成了order.*.create结果路由键order.create.success死活匹配不上排查了半天才发现是*的位置错了。*只匹配一个单词order.*.create要求第二段是任意单词、第三段必须是 create而实际第三段是 success。这种错误管理后台看不出来只能靠日志和测试。第二个坑是手动 ack 忘了调。有一次业务逻辑里加了个提前 return结果 ack 那行没执行到消息一直处于 unacked 状态队列越积越多。后来我养成了习惯把 ack 放在 finally 块里或者用 AOP 统一处理确认逻辑避免遗漏。第三个坑是消费者方法签名不匹配。RabbitListener标的方法参数类型必须和消息体对得上。我一开始用String接收但发送端发的是对象结果反序列化直接报错。后来统一用对象接收并且确保发送端和消费端的MessageConverter一致问题才解决。几条实在建议路由键的常量统一定义在一个类里发送端和消费端都引用它避免手写字符串出错上线前用管理后台的 “Publish message” 功能手动发几条测试消息比写单元测试还快给每个队列配监控告警Ready 数量持续增长就说明消费跟不上了得赶紧扩容或者排查。这套“Topic 交换机 多队列绑定 多接收器”的组合我在好几个项目里都用过从订单分发到日志收集都挺稳。核心就是把路由键设计好把 ack 控制好剩下的交给 RabbitMQ 就行。如果你在绑定规则上还有拿不准的地方建议直接在管理后台的 Exchanges 页面用 “Bindings” 功能试几下通配符匹配结果一目了然比看文档快得多。 
长河君 · 2026/10/20 12:09:15 · 阅读 1.2k SpringBoot整合RabbitMQ Topic交换机实现一发多收
1. 从一条消息说起为什么需要“一发多收”做过消息队列开发的人大概率都遇到过这样的场景用户下单成功后系统需要同时做几件事——给用户发短信、给仓库推发货单、给数据分析平台记一笔流水。最直觉的做法是在下单逻辑里依次调用这三个服务但这样一来下单接口的响应时间就等于三个服务的耗时之和任何一个环节卡住整个下单流程都得跟着等。更麻烦的是如果短信服务临时挂了你是让下单失败还是吞掉异常吞掉异常用户收不到短信不吞异常用户下不了单两头为难。消息队列就是来解决这类问题的。生产者只管把“订单已创建”这条消息丢出去至于谁来消费、消费几条、消费失败怎么办统统不用操心。而RabbitMQ 的 Topic 交换机正是实现“一条消息发往多个队列”的核心武器。它的工作模型不复杂生产者把消息发给交换机交换机根据路由键和绑定关系的匹配规则把同一条消息复制分发到多个队列里每个队列背后可以挂不同的消费者各干各的活互不干扰。这篇内容要聊的就是怎么在 SpringBoot 项目里把这套机制落地。具体来说我会带你走一遍完整流程从交换机、队列、绑定的声明到用注解方式写多个接收器再到路由键的设计技巧和实际踩过的坑。适合已经了解 RabbitMQ 基本概念、想在 SpringBoot 里把“一发多收”真正用起来的开发者。如果你还没接触过 RabbitMQ建议先补一下生产者、消费者、交换机这几个基础概念不然直接看绑定规则会有点懵。需要提前说明的是标题里提到的“接收器”在 SpringBoot 里通常有两种写法一种是RabbitListener注解直接标在方法上另一种是手动注册MessageListener容器。前者写起来清爽后者灵活度更高。我会以注解方式为主线因为它更符合 SpringBoot 的开发习惯同时在关键位置补充手动配置的替代方案方便你根据项目情况取舍。2. 核心机制拆解Topic 交换机到底怎么“一对多”2.1 交换机、队列、绑定三者的关系要理解“一条消息发往多个队列”得先把 RabbitMQ 里三个核心角色的分工理清楚。很多人初学时容易把交换机和队列混为一谈其实它们职责完全不同。交换机Exchange是消息进入 RabbitMQ 后的第一站它本身不存储消息只负责根据规则做转发。你可以把它想象成邮局的分拣中心信件到了分拣中心工作人员看地址决定投到哪个片区的邮筒。队列Queue才是真正存消息的地方消息在队列里排队等待消费者取走。绑定Binding则是交换机和队列之间的“路由规则”它告诉交换机什么样的消息该送到哪个队列。Topic 交换机的绑定规则有个特点路由键支持通配符匹配。这里有两个符号必须记牢*匹配一个单词以点分隔的一段#匹配零个或多个单词举个例子路由键是order.create.success那么绑定键order.*.success能匹配上*吃掉中间的createorder.#也能匹配上#吃掉后面所有但order.*匹配不上因为*只能吃一个单词而后面还剩create.success两段。正是这个通配符机制让同一条消息可以被多个不同绑定键的队列同时命中。比如一条路由键为order.create.success的消息队列 A 绑定了order.#队列 B 绑定了order.*.success队列 C 绑定了#.success那么这条消息会被复制三份分别进入 A、B、C 三个队列。这就是“一发多收”的底层原理。2.2 为什么选 Topic 而不是 Direct 或 FanoutRabbitMQ 有四种交换机类型选哪种直接决定了你的路由能力上限。我简单对比一下你就明白为什么这个场景非 Topic 莫属。交换机类型路由规则适用场景一对多能力Direct路由键完全相等才匹配点对点精确投递支持但需要多个绑定键完全相同Fanout忽略路由键广播到所有绑定队列全量广播支持但无法做筛选Topic通配符模糊匹配按业务维度灵活分发支持且可精细控制Headers按消息头匹配复杂条件匹配支持但性能较差、用得少Direct 交换机其实也能实现一对多只要把多个队列绑定到同一个路由键就行。但它的短板在于“不够灵活”——一旦业务需要按不同维度筛选你就得为每种组合单独建绑定维护成本飙升。Fanout 倒是简单粗暴所有队列都能收到可它完全没法筛选短信队列和日志队列收到的是同样的消息这显然不合理。Topic 的好处在于它用一套通配符规则覆盖了从精确到模糊的各种匹配需求。你可以用order.create精确投递也可以用order.#接收所有订单相关消息还能用#.error接收所有错误日志。一套机制多种玩法这就是它在实际项目里出场率最高的原因。2.3 消息复制的代价一对多不是没有成本这里必须泼一盆冷水。很多人以为“一条消息发往多个队列”是零成本的实际上 RabbitMQ 在背后做了消息复制。交换机每命中一个队列就会把消息内容写一份到那个队列里。三个队列就是三份存储十个队列就是十份。这意味着两件事。第一磁盘和内存开销随队列数量线性增长。如果你的消息体很大比如带附件、带大 JSON又绑定了很多队列broker 的压力会明显上升。第二消费速度受最慢的队列拖累。虽然各队列独立消费但如果某个队列积压严重它会持续占用 broker 资源间接影响其他队列的表现。所以我的经验是绑定到同一个交换机的队列数量要克制通常控制在个位数。如果发现某个业务需要广播给几十个消费者那可能不是消息队列该干的事考虑换成事件总线或者专门的推送系统更合适。3. SpringBoot 环境搭建与基础配置3.1 依赖引入与连接参数SpringBoot 集成 RabbitMQ 靠的是spring-boot-starter-amqp这个 starter它把 RabbitMQ 的 Java 客户端和 Spring 的抽象层都打包好了。在pom.xml里加这么一段dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency版本不用自己指定跟着 SpringBoot 父工程的版本管理走就行。如果你用的是 Gradle对应写法是implementation org.springframework.boot:spring-boot-starter-amqp。接下来是连接配置写在application.yml里spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest virtual-host: / listener: simple: acknowledge-mode: manual prefetch: 1 concurrency: 2 max-concurrency: 5这里有几个参数值得展开说。acknowledge-mode设成manual表示手动确认消费者处理完业务逻辑后自己调basicAck这样能避免消息丢失。如果你图省事用auto消息一投递出去就被标记为已消费业务处理失败也不会重投风险很大。prefetch设成 1 表示每个消费者一次只预取一条消息处理完再取下一条这样能保证负载均衡不会出现某个消费者囤了一堆消息而其他消费者闲着的情况。concurrency和max-concurrency控制消费者线程数的下限和上限根据你的业务耗时和机器配置调整。提示guest账号默认只允许本机连接生产环境务必创建独立账号并设置合理权限同时开启虚拟主机隔离不同业务。3.2 声明交换机、队列与绑定SpringBoot 里声明这些组件有两种方式用Bean在配置类里声明或者用RabbitListener注解自动声明。我推荐前者因为集中管理更清晰排查问题时一眼能看到全貌。Configuration public class RabbitConfig { public static final String TOPIC_EXCHANGE order.topic.exchange; public static final String SMS_QUEUE order.sms.queue; public static final String WAREHOUSE_QUEUE order.warehouse.queue; public static final String ANALYTICS_QUEUE order.analytics.queue; Bean public TopicExchange orderExchange() { return new TopicExchange(TOPIC_EXCHANGE, true, false); } Bean public Queue smsQueue() { return new Queue(SMS_QUEUE, true); } Bean public Queue warehouseQueue() { return new Queue(WAREHOUSE_QUEUE, true); } Bean public Queue analyticsQueue() { return new Queue(ANALYTICS_QUEUE, true); } Bean public Binding smsBinding() { return BindingBuilder.bind(smsQueue()) .to(orderExchange()) .with(order.#); } Bean public Binding warehouseBinding() { return BindingBuilder.bind(warehouseQueue()) .to(orderExchange()) .with(order.create.*); } Bean public Binding analyticsBinding() { return BindingBuilder.bind(analyticsQueue()) .to(orderExchange()) .with(#.success); } }这段代码里TopicExchange的第二个参数true表示持久化broker 重启后交换机还在第三个参数false表示不自动删除。队列的true同样表示持久化。绑定关系用BindingBuilder链式构建.with()后面跟的就是绑定键。三个队列的绑定键设计是有讲究的。短信队列用order.#意思是所有订单相关消息它都要因为用户可能对下单、取消、退款都关心。仓库队列用order.create.*只关心创建类消息取消和退款跟它无关。分析队列用#.success只收成功类事件失败的不进分析库。这样一条order.create.success的消息三个队列全都能命中各取所需。3.3 消息发送端的写法发送端用RabbitTemplate就够了SpringBoot 会自动注入。核心代码如下Service public class OrderService { Autowired private RabbitTemplate rabbitTemplate; public void createOrder(Order order) { // 业务逻辑省略 String routingKey order.create.success; rabbitTemplate.convertAndSend( RabbitConfig.TOPIC_EXCHANGE, routingKey, order, message - { message.getMessageProperties() .setMessageId(UUID.randomUUID().toString()); message.getMessageProperties() .setContentType(application/json); return message; } ); } }convertAndSend的四个参数分别是交换机名、路由键、消息体、消息后置处理器。后置处理器里我设置了消息 ID 和内容类型消息 ID 用于幂等去重内容类型告诉消费者怎么反序列化。如果你发送的是对象SpringBoot 默认用 JDK 序列化但 JDK 序列化跨语言兼容性差建议换成 JacksonBean public MessageConverter jsonMessageConverter() { return new Jackson2JsonMessageConverter(); }把这个 Bean 注册进去RabbitTemplate和监听器都会自动用它做序列化省心。4. 多个接收器的实现与路由键设计4.1 用 RabbitListener 写三个消费者接收器的核心就是RabbitListener注解标在方法上指定监听的队列名。三个队列对应三个方法各写各的业务逻辑Component public class OrderMessageReceiver { RabbitListener(queues RabbitConfig.SMS_QUEUE) public void handleSms(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(短信服务收到订单 order.getOrderNo()); // 调用短信发送逻辑 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); } } RabbitListener(queues RabbitConfig.WAREHOUSE_QUEUE) public void handleWarehouse(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(仓库服务收到订单 order.getOrderNo()); // 生成发货单 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, true); } } RabbitListener(queues RabbitConfig.ANALYTICS_QUEUE) public void handleAnalytics(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { System.out.println(分析服务收到订单 order.getOrderNo()); // 写入分析库 channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); } } }每个方法都接收Channel和deliveryTag这是手动确认的前提。basicAck表示确认消费basicNack的第三个参数表示是否重新入队。注意仓库那个方法我传了true因为发货单生成失败通常是临时问题比如数据库连接抖动重试有意义而短信和分析失败传false避免无限重试把队列堵死。注意手动确认模式下如果消费者方法抛异常且你没捕获SpringBoot 默认会拒绝消息并重新入队可能导致死循环。所以务必用 try-catch 包住业务逻辑明确控制 ack 和 nack。4.2 路由键设计的三个实用原则路由键设计得好不好直接决定了后期扩展是轻松还是痛苦。我总结了三条原则都是从实际项目里踩出来的。第一用点分层级从粗到细。比如order.create.success第一段是业务域第二段是动作第三段是结果。这样绑定的时候可以按任意层级筛选。反过来写成createOrderSuccess就没法拆了只能全匹配。第二预留扩展位。我习惯在业务域后面留一段表示来源或渠道比如order.app.create.success、order.web.create.success。这样以后想单独处理 App 渠道的订单绑定order.app.#就行不用改生产者代码。第三避免用#开头做绑定。虽然#.success语法上合法但它会匹配所有以 success 结尾的消息包括其他业务域的。如果交换机是共享的很容易误收。建议绑定键至少带上业务域前缀比如order.#.success。4.3 一个消费者监听多个队列的写法有时候几个队列的处理逻辑高度相似没必要写多个方法。RabbitListener的queues属性支持数组可以一次监听多个RabbitListener(queues {RabbitConfig.SMS_QUEUE, RabbitConfig.ANALYTICS_QUEUE}) public void handleMultiple(Order order, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { // 根据消息来源做不同处理 channel.basicAck(tag, false); }但这里有个坑多个队列共用一个消费者方法时你没法直接知道消息是从哪个队列来的。解决办法是从消息头里取AmqpHeaders.CONSUMER_QUEUE或者干脆拆成两个方法。我的建议是除非逻辑真的完全一样否则还是分开写可读性更好排查问题也方便。5. 实操验证与常见问题排查5.1 完整验证流程配置写完后怎么确认“一发多收”真的生效了我通常按这个顺序验证。第一步启动项目去 RabbitMQ 管理后台默认 15672 端口看 Exchanges 和 Queues 页面确认交换机和三个队列都创建成功绑定关系也在 Bindings 里能看到。如果队列没出现多半是配置类没被扫描到检查一下包路径。第二步调一次下单接口观察控制台。三个接收器方法应该都打印出订单号说明消息被复制分发到了三个队列。如果只有部分打印去管理后台看对应队列的 Ready 数量有堆积说明消费者没起来或者绑定键写错了。第三步故意在某个接收器里抛异常看消息是否按预期 nack。比如让短信方法抛异常并basicNack(tag, false, false)然后去后台看这个队列的消息是不是被丢弃了或者进了死信队列如果你配了的话而其他两个队列不受影响。这一步能验证队列之间的隔离性。5.2 常见问题速查表现象可能原因排查方向消息只进了一个队列绑定键不匹配检查路由键和绑定键的通配符是否对得上三个队列都收不到消息交换机名或路由键写错对比发送端和配置类的常量消费者方法不执行队列没绑定或监听器没注册看后台队列的 Consumers 数量是否为 0消息重复消费ack 时机不对或网络抖动引入消息 ID 做幂等检查 ack 是否在业务后启动报连接拒绝broker 没启动或端口不对确认 5672 端口可达账号密码正确消息体反序列化失败序列化方式不一致统一用 Jackson2JsonMessageConverter5.3 几个容易忽略的细节关于消息顺序。一条消息复制到三个队列后三个队列的消费是并行的谁先处理完不确定。如果你有严格的顺序要求比如必须先记流水再发短信那不能靠队列来保证得在业务层加状态机或者用单队列串行消费。关于死信队列。生产环境强烈建议给每个业务队列配死信队列。消息被 nack 且不重新入队时会进死信队列方便事后排查。配置方式是在声明队列时加参数Bean public Queue smsQueue() { return QueueBuilder.durable(SMS_QUEUE) .withArgument(x-dead-letter-exchange, order.dlx.exchange) .withArgument(x-dead-letter-routing-key, order.sms.dead) .build(); }关于消费者线程数。concurrency不是越大越好。如果业务是 CPU 密集型线程数超过 CPU 核心数反而增加上下文切换开销如果是 IO 密集型可以适当调大。我一般从 2 开始压测后根据吞吐量调整。关于消息体大小。RabbitMQ 默认单条消息最大 128MB但实际别发这么大的。超过 1MB 的消息就会明显影响性能建议把大内容存到对象存储消息里只传引用地址。6. 我踩过的坑和几条实在建议说几个我在实际项目里真金白银换来的教训。第一个坑是绑定键写反了通配符位置。有次我把order.create.*写成了order.*.create结果路由键order.create.success死活匹配不上排查了半天才发现是*的位置错了。*只匹配一个单词order.*.create要求第二段是任意单词、第三段必须是 create而实际第三段是 success。这种错误管理后台看不出来只能靠日志和测试。第二个坑是手动 ack 忘了调。有一次业务逻辑里加了个提前 return结果 ack 那行没执行到消息一直处于 unacked 状态队列越积越多。后来我养成了习惯把 ack 放在 finally 块里或者用 AOP 统一处理确认逻辑避免遗漏。第三个坑是消费者方法签名不匹配。RabbitListener标的方法参数类型必须和消息体对得上。我一开始用String接收但发送端发的是对象结果反序列化直接报错。后来统一用对象接收并且确保发送端和消费端的MessageConverter一致问题才解决。几条实在建议路由键的常量统一定义在一个类里发送端和消费端都引用它避免手写字符串出错上线前用管理后台的 “Publish message” 功能手动发几条测试消息比写单元测试还快给每个队列配监控告警Ready 数量持续增长就说明消费跟不上了得赶紧扩容或者排查。这套“Topic 交换机 多队列绑定 多接收器”的组合我在好几个项目里都用过从订单分发到日志收集都挺稳。核心就是把路由键设计好把 ack 控制好剩下的交给 RabbitMQ 就行。如果你在绑定规则上还有拿不准的地方建议直接在管理后台的 Exchanges 页面用 “Bindings” 功能试几下通配符匹配结果一目了然比看文档快得多。