1. 为什么需要理解RabbitMQ的工作模式很多刚接触消息队列的开发者对RabbitMQ的第一印象就是生产者发消息消费者收消息觉得这东西简单得很。但真正到了业务场景里你会发现事情远没有这么简单——有的消息需要所有下游系统都收到有的消息只需要其中一个节点处理就行有的消息要按优先级分发还有的消息需要根据不同的路由规则送到不同的队列里。这些不同的需求对应的就是RabbitMQ的不同工作模式。我自己在几个项目里都用到了RabbitMQ从最开始只会用最简单的队列模式到后来逐渐把发布订阅、路由、主题等模式都用了一遍踩过的坑确实不少。这篇文章就把RabbitMQ的四种核心工作模式彻底讲清楚包括它们各自解决什么问题、底层是怎么实现的、代码怎么写、实际用的时候要注意什么。不管你是刚接触消息队列的新手还是已经用过一段时间但没系统梳理过的开发者应该都能从中找到有用的东西。RabbitMQ的工作模式本质上是在回答一个问题一条消息从生产者发出后应该以什么样的规则投递到一个或多个队列最终被一个或多个消费者处理不同的投递规则就构成了不同的工作模式。理解这些模式的关键不在于记住代码模板而在于搞清楚每种模式背后的交换机类型、路由键机制和队列绑定关系。下面我会按照从简单到复杂的顺序逐一拆解这四种模式每种模式都会配上完整的代码示例、参数说明和实际使用中的经验教训。2. 四种工作模式的整体设计与核心概念2.1 先搞清楚RabbitMQ的核心组件在展开讲四种模式之前有必要先把RabbitMQ的几个核心概念理清楚不然后面讲模式的时候容易懵。你可以把RabbitMQ想象成一个邮局系统生产者Producer寄信的人负责把消息发送到RabbitMQ。交换机Exchange邮局的分拣中心它不存储消息只负责根据规则把消息转发到对应的队列。队列Queue邮局的信箱消息真正存储的地方消费者从这里取消息。消费者Consumer收信的人从队列里取消息并处理。路由键Routing Key寄信时写在信封上的地址信息交换机根据它来决定消息往哪个队列送。绑定Binding队列和交换机之间的关联关系相当于告诉分拣中心这个信箱归你管。这六个概念贯穿所有工作模式区别只在于交换机的类型不同、路由键的使用方式不同、队列和消费者的数量不同。2.2 四种模式的对比与选型思路RabbitMQ常用的四种工作模式分别是简单队列模式Simple Queue、工作队列模式Work Queue、发布订阅模式Publish/Subscribe、路由模式Routing。如果再加上主题模式Topic其实是五种但主题模式可以看作是路由模式的模糊匹配版本很多资料把它归入路由模式的延伸这里我把它单独拿出来讲因为它在实际项目中的使用频率非常高。先上一张对比表让你对四种模式有个全局印象模式名称交换机类型队列数量消费者数量消息投递规则典型场景简单队列无默认交换机11一对一单点异步处理工作队列无默认交换机1多个竞争消费一条消息只被一个消费者处理任务分发、负载均衡发布订阅Fanout多个多个广播所有绑定队列都收到通知广播、缓存刷新路由模式Direct多个多个按路由键精确匹配分类处理、定向分发主题模式Topic多个多个按通配符模糊匹配多维度订阅、日志分级选型的核心逻辑其实就一句话先问自己这条消息需要被几个消费者处理再问需不需要按规则筛选消费者。如果只需要一个消费者处理用简单队列或工作队列如果需要多个消费者都收到用发布订阅如果需要按条件筛选哪些消费者收到用路由或主题模式。2.3 交换机类型决定了模式的行为很多人学RabbitMQ的时候容易混淆模式和交换机类型这两个概念。其实模式是由交换机类型和队列绑定方式共同决定的。RabbitMQ有四种交换机类型Default Exchange默认交换机每个队列创建时会自动绑定到一个同名的默认交换机路由键就是队列名。简单队列和工作队列用的就是它。Fanout Exchange不看路由键把消息广播给所有绑定的队列。发布订阅模式用它。Direct Exchange根据路由键精确匹配只有路由键完全一致的队列才会收到消息。路由模式用它。Topic Exchange根据路由键的模式匹配支持星号和井号通配符。主题模式用它。理解了这四种交换机的行为差异四种工作模式的区别也就一目了然了。3. 简单队列与工作队列模式的实操要点3.1 简单队列模式最基础的一对一通信简单队列模式是RabbitMQ里最原始的使用方式。一个生产者一个队列一个消费者消息从生产者发出后进入队列消费者从队列里取出来处理。就这么简单。它的代码结构大概是这样的以Java为例其他语言逻辑类似// 生产者 public class SimpleProducer { public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); factory.setPort(5672); factory.setUsername(guest); factory.setPassword(guest); try (Connection connection factory.newConnection(); Channel channel connection.createChannel()) { // 声明队列参数依次为队列名、是否持久化、是否独占、是否自动删除、其他参数 channel.queueDeclare(simple_queue, false, false, false, null); String message 这是一条简单队列消息; // 使用默认交换机路由键就是队列名 channel.basicPublish(, simple_queue, null, message.getBytes(UTF-8)); System.out.println(消息已发送 message); } } }// 消费者 public class SimpleConsumer { public static void main(String[] args) throws Exception { ConnectionFactory factory new ConnectionFactory(); factory.setHost(localhost); Connection connection factory.newConnection(); Channel channel connection.createChannel(); channel.queueDeclare(simple_queue, false, false, false, null); // 关闭自动确认改为手动确认 channel.basicConsume(simple_queue, false, (consumerTag, delivery) - { String message new String(delivery.getBody(), UTF-8); System.out.println(收到消息 message); // 手动确认 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }, consumerTag - {}); } }这段代码里有个关键点需要特别注意basicConsume的第二个参数是autoAck自动确认。我见过太多新手直接传true结果消费者处理消息失败时消息就丢了。正确的做法是传false然后在业务逻辑处理成功后手动调用basicAck。如果处理失败可以调用basicNack让消息重新入队或者丢弃。注意简单队列模式在实际生产环境中用得并不多因为它没有负载均衡能力消费者一旦挂掉消息就会堆积。它更适合用来做本地测试或者理解RabbitMQ的基本原理。3.2 工作队列模式让多个消费者分担压力工作队列模式也叫任务队列模式在简单队列的基础上增加了多个消费者。多个消费者同时监听同一个队列RabbitMQ会把消息轮流分发给不同的消费者每条消息只会被一个消费者处理。这就是所谓的竞争消费。这个模式解决的核心问题是当消息生产速度远大于单个消费者的处理速度时通过增加消费者来提升整体吞吐量。默认情况下RabbitMQ采用轮询Round-Robin的方式分发消息即消费者1一条、消费者2一条、消费者1一条……依次循环。但这里有个坑如果消费者1处理消息很快消费者2处理很慢轮询会导致消费者1空闲而消费者2积压。解决办法是设置basicQos的prefetchCount参数// 每次只预取一条消息处理完确认后再取下一跳 channel.basicQos(1);这个参数的含义是消费者在未确认当前消息之前最多还能预取多少条消息。设置为1就实现了能者多劳的效果——处理快的消费者自然拿到更多消息。我在一个订单处理系统里用过这个模式当时有3个消费者实例其中一个部署在配置较低的机器上。没设basicQos之前那个慢节点经常积压几十条消息设了之后基本均衡了。这个参数看起来不起眼但在工作队列模式里几乎是必调的。3.3 消息确认机制别让消息悄悄丢失不管是简单队列还是工作队列消息确认机制都是绕不开的话题。RabbitMQ的消息确认分两种自动确认autoAcktrue消息一发给消费者就算确认了不管消费者有没有处理成功。优点是快缺点是消费者崩溃时消息就丢了。手动确认autoAckfalse消费者处理完后显式调用basicAckRabbitMQ才把消息从队列里删除。如果消费者崩溃且没有确认消息会重新入队。手动确认有三种操作// 确认成功 channel.basicAck(deliveryTag, false); // 拒绝并重新入队 channel.basicNack(deliveryTag, false, true); // 拒绝并丢弃或进入死信队列 channel.basicNack(deliveryTag, false, false);实操心得手动确认一定要放在业务逻辑的try-catch里处理成功才ack处理失败根据情况nack。我见过有人把ack写在业务逻辑外面结果业务抛异常了消息还是被确认了数据直接丢了。另外如果消息反复处理失败不要无限重新入队应该设置重试次数超过后转入死信队列否则会形成死循环。4. 发布订阅与路由模式的深度解析4.1 发布订阅模式一条消息广播给所有订阅者发布订阅模式的核心是Fanout交换机。生产者把消息发给Fanout交换机交换机会把消息复制多份广播给所有绑定到它的队列。每个队列都有自己的消费者所以一条消息会被多个消费者分别处理。这个模式的典型应用场景是一个事件发生后需要触发多个不相关的下游动作。比如用户注册成功后需要同时发送欢迎邮件、初始化用户积分、记录行为日志。这三个动作互相独立用发布订阅模式就很合适。代码实现的关键在于声明Fanout交换机并绑定队列// 生产者 channel.exchangeDeclare(fanout_exchange, BuiltinExchangeType.FANOUT); String message 用户注册成功事件; channel.basicPublish(fanout_exchange, , null, message.getBytes(UTF-8));// 消费者1邮件服务 channel.exchangeDeclare(fanout_exchange, BuiltinExchangeType.FANOUT); String queueName channel.queueDeclare().getQueue(); // 创建临时队列 channel.queueBind(queueName, fanout_exchange, ); channel.basicConsume(queueName, false, (consumerTag, delivery) - { // 发送欢迎邮件 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); }, consumerTag - {});注意这里用了channel.queueDeclare()的无参版本它会创建一个临时队列随机名称、非持久化、独占、自动删除。这种队列在消费者断开连接后会自动删除非常适合发布订阅模式——因为每个订阅者通常只需要在运行期间接收消息。注意Fanout交换机完全忽略路由键你传什么路由键它都不看直接广播给所有绑定的队列。所以绑定的时候路由键传空字符串就行。4.2 路由模式按路由键精确分发路由模式用的是Direct交换机。它和发布订阅的区别在于发布订阅是无差别广播路由模式则会根据路由键进行精确匹配只有路由键完全一致的队列才会收到消息。举个例子一个日志系统需要把不同级别的日志送到不同的处理队列。error级别的日志送到错误处理队列info级别的送到普通记录队列warning级别的两个队列都要收到。用路由模式可以这样绑定// 生产者发送error日志 channel.exchangeDeclare(log_exchange, BuiltinExchangeType.DIRECT); channel.basicPublish(log_exchange, error, null, 错误日志内容.getBytes()); // 消费者1错误处理队列绑定error和warning channel.queueBind(error_queue, log_exchange, error); channel.queueBind(error_queue, log_exchange, warning); // 消费者2普通记录队列绑定info和warning channel.queueBind(info_queue, log_exchange, info); channel.queueBind(info_queue, log_exchange, warning);这样一条error日志只会进error_queue一条info日志只会进info_queue而一条warning日志两个队列都会进。这就是路由模式的灵活性所在——同一个队列可以绑定多个路由键同一个路由键也可以绑定多个队列。路由模式在实际项目中的使用频率很高尤其是需要按业务类型、按地区、按优先级等维度分发消息的场景。它的路由规则简单直接不容易出错性能也很好。4.3 主题模式通配符匹配的进阶路由主题模式用的是Topic交换机它是路由模式的增强版。区别在于Topic交换机的路由键支持两种通配符星号*匹配一个单词。井号#匹配零个或多个单词。路由键的单词之间用点号分隔比如order.create.success、user.login.fail。假设有一个电商系统需要按业务模块和操作类型订阅消息// 生产者发送订单创建消息 channel.exchangeDeclare(topic_exchange, BuiltinExchangeType.TOPIC); channel.basicPublish(topic_exchange, order.create.success, null, 订单创建成功.getBytes()); // 消费者1订阅所有订单相关消息 channel.queueBind(order_queue, topic_exchange, order.#); // 消费者2订阅所有模块的创建成功消息 channel.queueBind(create_queue, topic_exchange, *.create.*); // 消费者3订阅所有成功消息 channel.queueBind(success_queue, topic_exchange, *.*.success);这样order.create.success这条消息会被三个队列都收到因为order.#匹配order.create.success井号匹配create.success两个单词*.create.*匹配order.create.success两个星号分别匹配order和success*.*.success匹配order.create.success两个星号分别匹配order和create主题模式的灵活性最高但也最容易出错。我踩过的一个坑是井号匹配零个或多个单词所以order.#也能匹配order本身。如果你不希望匹配到order就得用order.*或者更精确的模式。这个细节在官方文档里写得很清楚但实际用的时候很容易忽略。实操心得主题模式的路由键设计要有前瞻性最好在项目初期就定好命名规范比如业务模块.操作类型.结果状态这样的三段式。后期如果路由键格式不统一通配符匹配会变得非常混乱。另外Topic交换机的性能比Direct略低因为需要做模式匹配但在大多数场景下这个差异可以忽略。5. 常见问题与排查技巧实录5.1 消息丢失的三种情况和排查方法消息丢失是RabbitMQ使用中最常见的问题可能发生在三个环节丢失环节原因解决方案生产者到交换机网络中断、交换机不存在开启生产者确认Publisher Confirm交换机到队列路由键不匹配、队列不存在设置mandatory标志Return回调队列到消费者消费者崩溃、自动确认手动确认队列持久化消息持久化生产者确认的代码示例channel.confirmSelect(); // 开启确认模式 channel.basicPublish(exchange, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, // 消息持久化 message.getBytes()); // 同步等待确认 if (channel.waitForConfirms()) { System.out.println(消息发送成功); } else { System.out.println(消息发送失败需要重试); }队列持久化和消息持久化要同时设置才有效queueDeclare的第二个参数设为true消息的deliveryMode设为2MessageProperties.PERSISTENT_TEXT_PLAIN就是这个值。只设其中一个重启后消息还是会丢。5.2 消息重复消费的原因与幂等处理消息重复消费在RabbitMQ里几乎无法完全避免因为网络抖动、消费者超时等都可能导致消息重新入队。解决思路不是杜绝重复而是让重复消费不影响业务也就是幂等处理。常见的幂等方案有数据库唯一索引给消息ID建唯一索引重复插入会报错捕获异常即可。Redis去重用消息ID作为key处理前先查Redis处理完写入Redis并设置过期时间。状态机判断业务状态只能从A流转到B重复消息到达时状态已经是B了直接忽略。我在一个支付回调系统里用的是Redis去重方案key的过期时间设为24小时。实测下来重复消息的比例大概在万分之几但如果没有幂等处理这万分之几就会造成重复扣款后果很严重。5.3 队列积压的应急处理与预防队列积压通常是因为消费者处理速度跟不上生产速度。应急处理的办法有临时增加消费者实例最快的方式但要注意basicQos的设置否则新实例可能抢不到消息。批量消费用basicConsume的批量确认模式一次取多条消息处理减少网络往返。降级处理把非核心消息先丢弃或转入死信队列优先保证核心消息的处理。预防积压的根本办法是做好容量规划。我的经验是消费者数量 峰值QPS × 平均处理耗时 / 单实例并发数然后留30%的余量。比如峰值QPS是1000平均处理耗时50ms单实例并发10那么需要的消费者实例数大约是1000×0.05/105个再加30%余量就是7个。5.4 交换机与队列绑定不生效的排查步骤绑定不生效是新手最容易遇到的问题排查顺序如下确认交换机存在在管理界面查看Exchanges列表或者用channel.exchangeDeclarePassive(exchange_name)检查。确认队列存在同样用queueDeclarePassive检查。确认绑定关系存在在管理界面的交换机详情页查看Bindings或者用channel.exchangeBind重新绑定。确认路由键匹配Direct交换机要求完全一致Topic交换机要检查通配符是否写对。确认消息确实发到了交换机用basicPublish的mandatory标志如果消息无法路由会触发Return回调。实操心得RabbitMQ的管理界面默认端口15672是排查问题的利器Exchanges、Queues、Bindings三个页面基本能定位90%的问题。建议开发阶段就打开管理界面边操作边观察队列和绑定的变化比看日志直观得多。6. 模式选型的实战经验与性能考量6.1 从业务需求反推模式选型选模式不要从技术出发要从业务出发。我通常问三个问题第一个问题这条消息需要被几个消费者处理一个就用简单队列或工作队列多个就用发布订阅、路由或主题。第二个问题多个消费者是全部收到还是按条件筛选全部收到用发布订阅按条件筛选用路由或主题。第三个问题筛选条件是精确值还是模式匹配精确值用路由模式匹配用主题。这三个问题回答完模式基本就定了。比如订单创建成功后需要通知库存服务、积分服务、通知服务且这三个服务都要收到答案就是发布订阅模式。6.2 性能对比与压测数据参考我在一台4核8G的机器上做过简单的压测四种模式的吞吐量大致如下消息体1KB消费者手动确认模式生产者TPS消费者TPS备注简单队列约12000约8000单消费者工作队列约12000约250003个消费者发布订阅约10000约200003个队列各1消费者路由模式约11000约220003个队列各1消费者主题模式约9000约180003个队列各1消费者这些数据只是参考实际性能受消息大小、网络延迟、磁盘IO等因素影响很大。但可以看出一个规律Fanout和Direct的性能接近Topic因为要做模式匹配性能略低10%左右。在大多数业务场景下这个差异完全可以接受。6.3 持久化对性能的影响与取舍持久化是保证消息不丢的关键但会显著降低吞吐量。我实测过开启队列持久化消息持久化后生产者TPS从12000降到了约4000降幅超过60%。所以持久化不是无脑开的要根据业务场景取舍必须持久化订单、支付、库存等涉及资金的场景。可以不持久化日志收集、实时监控、临时通知等允许少量丢失的场景。折中方案队列持久化但消息不持久化或者用镜像队列仲裁队列提升可靠性。注意即使开了持久化消息也不是100%不丢。RabbitMQ的持久化是异步刷盘的在极端情况下如机器突然断电仍可能丢失少量消息。如果业务要求绝对可靠需要在应用层做补偿比如本地消息表定时对账。6.4 集群环境下的模式表现差异在RabbitMQ集群中队列的元数据会在所有节点同步但消息实体只存在于队列所在的节点。这意味着简单队列和工作队列如果队列所在节点挂了消息就丢了除非用了镜像队列或仲裁队列。发布订阅和路由模式每个队列独立存储一个队列所在节点挂了不影响其他队列。主题模式同路由模式。所以在集群环境下发布订阅、路由、主题模式的高可用性天然优于简单队列和工作队列因为消息分散在多个队列里。如果业务对可靠性要求高建议优先选择多队列的模式并配合镜像队列或仲裁队列使用。7. 我个人的一些使用体会RabbitMQ的四种工作模式说到底是在解决消息如何路由这一个问题。简单队列和工作队列用默认交换机消息直接进队列发布订阅用Fanout交换机消息广播到所有队列路由模式用Direct交换机按路由键精确匹配主题模式用Topic交换机按通配符模糊匹配。把这四种交换机的行为搞清楚四种模式就都通了。我在实际项目里用得最多的是工作队列和主题模式。工作队列用来做任务分发简单可靠主题模式用来做事件通知灵活性强。发布订阅和路由模式用得相对少一些因为它们的场景往往可以用主题模式替代而主题模式的扩展性更好。最后分享一个小技巧在项目初期就把交换机和队列的命名规范定好比如业务模块.消息类型的格式交换机用xxx.exchange队列用xxx.queue。后期队列多了以后规范的命名能省下大量排查时间。我见过一个项目队列名全是queue1、queue2这种运维的时候完全不知道哪个队列对应哪个业务那叫一个痛苦。
