一篇标题为“RabbitMQ的六种工作方式及其应用场景与代码示例”的博文，通常就是准备给刚接触消息队列的人讲清楚：这玩意儿到底有几种玩法，每种玩法什么时候用。作为一个在项目里被 RabbitMQ 各种模式“教育”过的人，我想先从一个真实感受说起——网上讲 RabbitMQ 模式的教程很多，但大部分代码一跑就通，一换业务场景就不知道该选哪一种。原因很简单，死记了模式没理解模型。所以这篇文章我尽量用打工人的语言，把六种工作方式从头到尾掰开揉碎，每个模式配一段能直接跑的Python代码和具体的业务画面，帮你把“交换机、队列、绑定、路由键”这一串概念真正焊进脑子里。

1. 整体设计思路：为什么先搞懂交换机，才能看懂六种模式

RabbitMQ 之所以能玩出六种工作方式，不是因为队列本身多复杂，而是它中间夹了一层别人没有的“路由器”——交换机（Exchange）。生产者的消息从来不直接丢进队列，而是先扔给交换机，再由交换机按照预设规则分发给一个或多个队列。这条链路一旦想明白，后面五种模式全靠换不同“路由规则”来实现，读代码时会轻松很多。

1.1 核心概念串讲：消息到底经历了什么

想象你在公司前台放了一个收发室（交换机），所有快递（消息）到了之后由收发室大妈决定送到哪个部门信箱（队列）。大妈的送信规则有四种：

• 见人就发一份（Fanout广播）
• 看门牌号精确匹配再送（Direct直连）
• 按地址模糊匹配再送（Topic主题）
• 实在找不到就退货或直接烧掉（Default默认交换机 + 死信）

对应到 RabbitMQ 里，这些规则就是交换机的类型：fanout、direct、topic、headers（headers用得少，本文不展开）。所谓六种工作方式，本质上是“两种基本队列模型 + 四种交换机分发策略”的组合拳。

1.2 为什么六种模式是渐进式选择题

第一次接触时我也犯过迷糊：六种模式，是不是每个项目都得用到？其实不是。RabbitMQ 官方把这六种方式当作“由浅入深”的教学路径，而不是“必选套餐”。六种模式恰好覆盖了从最简单的点对点通信到复杂的RPC调用，它们之间的递进关系是：

• 简单队列：一个生产者一个消费者，练手级
• 工作队列：一个生产者多个消费者抢活干，解决耗时任务
• 发布订阅：一个生产者多个消费者，大家都得收到同一份消息
• 路由模式：按指定规则让部分消费者收消息，实现定向通知
• 主题模式：通过通配符做更灵活的消息筛选，支持复杂业务匹配
• RPC模式：用消息队列做同步请求-响应，满足跨系统调用

这篇博文也会按照这个顺序讲，因为每一步都在前一步基础上加东西，理解成本最低。你甚至可以理解为：项目里的需求是逐渐变复杂的，消息中间件的用法也必须跟着变复杂。

2. 简单队列与工作队列：从零开始，先让消息“动起来”

很多教程喜欢一上来就给你看完整工程，实际反而吓人。最简单的方式是你本地装好 RabbitMQ 后，用Python的pika库花十分钟跑通两个脚本。这两节先把地基打牢，我会把每个参数拆开讲为什么这么设。

2.1 简单队列：一个生产者，一个消费者

业务场景最典型的就是日志上报。你一个后台脚本定时收集监控数据，扔进队列里，另一边有个消费进程专门负责落库。多一个消费者反而没意义，因为日志处理本身就是顺序写盘，并发写反而要考虑锁问题。

生产者代码（send.py）：

import pika connection = pika.BlockingConnection( pika.ConnectionParameters( host='localhost', credentials=pika.PlainCredentials('guest', 'guest') ) ) channel = connection.channel() channel.queue_declare(queue='hello', durable=True) channel.basic_publish( exchange='', routing_key='hello', body='Hello RabbitMQ!', properties=pika.BasicProperties(delivery_mode=2) ) print("消息已发送") connection.close()

消费者代码（receive.py）：

import pika connection = pika.BlockingConnection( pika.ConnectionParameters(host='localhost') ) channel = connection.channel() channel.queue_declare(queue='hello', durable=True) def callback(ch, method, properties, body): print("收到消息:", body.decode()) ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_consume(queue='hello', on_message_callback=callback) print("等待消息中...") channel.start_consuming()

这里必须解释两个细节。第一，exchange=''代表使用默认交换机，此时routing_key直接写队列名，消息就能精准投递——很多新手以为默认交换机是“没有交换机”，其实它本质是一个 direct 类型的交换机，只是不需要显式声明。第二，delivery_mode=2表示消息持久化，queue_declare里的durable=True是队列持久化，两行结合起来才能保证 RabbitMQ 重启后消息不丢。

注意：queue_declare设置durable=True只能对新建队列生效。如果你之前已经用非持久化方式创建过同名队列，再声明持久化是无效的——必须先换一个队列名或者删除原队列。这个是我自己在测试环境里踩过的坑，改了代码半天没效果，查了半天发现是旧队列残留。

2.2 工作队列：一个生产者，多个消费者轮番干活

把业务升级一下：假设你不是处理日志，而是后台批量发送注册邮件，每封邮件要调第三方接口，耗时50-200毫秒。这时候单消费者肯定扛不住，标准解法就是引入多个消费者共用一个队列，RabbitMQ负责把消息自动分发出去。

生产者（task_new.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) message = ' '.join(sys.argv[1:]) or 'Hello World!' channel.basic_publish( exchange='', routing_key='task_queue', body=message.encode(), properties=pika.BasicProperties(delivery_mode=2) ) print("发送任务：", message) connection.close()

消费者（worker.py）：

import pika, time connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.queue_declare(queue='task_queue', durable=True) print("工作进程等待任务...") def callback(ch, method, properties, body): print("接收任务:", body.decode()) time.sleep(body.count(b'.')) # 模拟耗时 print("任务完成") ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(queue='task_queue', on_message_callback=callback) channel.start_consuming()

这套代码的关键就在最后两行：basic_qos(prefetch_count=1)。没有这行的话，RabbitMQ 默认是轮询分发——你一个我一个，完全不care谁手里积压了多少活。比如消费者A处理一条消息要10秒，消费者B处理一条只要1秒，默认轮询模式下A永远会被活活累死，B在那儿闲着等。

加上prefetch_count=1之后，语义变成“只要某个消费者手里还有未确认的消息，就别再给它发新的”，这样大部分消息会自动流向上一个处理完的消费者。这就是“能者多劳”的消息分发模型，也是工作队列模式的核心价值。

实操感悟：prefetch_count不是越大越好，也不是越小越好。如果消费者处理消息很快，prefetch=1会引入大量网络往返（每处理完一条就要向 Broker 取一条新消息）；如果消费者处理很慢，prefetch太大又会让积压集中。我的经验是先从prefetch=1起步，压测后逐步上调，一般内部系统2-10之间比较合适；对外部接口有限速要求的场景，永远保持1。

2.3 两种模式最容易踩的坑：确认机制与消息持久化

简单队列和工作队列虽然基础，但坑一点都不少。我见过无数次生产事故，归结起来就两类：

• 忘写basic_ack：消费者收到消息后直接处理，忘了确认（或者代码里确认逻辑被异常吞掉了），RabbitMQ 认为消息投递失败，重启后会再次投递，导致消息重复消费。反过来，如果你消费后立刻ack，但业务逻辑在后面崩了，消息就永久丢失。最稳妥的姿势是先处理完业务，最后再调ack，两害相权求得最小损失。
• 只设了队列持久化，没设消息持久化：queue_declare参数里的durable=True只代表队列本身重启后还在，不代表队列里的消息能重建。要保证消息不丢，basic_publish里必须带properties=pika.BasicProperties(delivery_mode=2)。两边都做了，才算真正实现了“消息不丢失”。

如果你还希望消费者挂了之后未被确认的消息重新入队，可以给消费者设置basic_qos之外再加一个手动basic_nack并指定requeue=True，这样可以做出“处理失败后自动重新入队重试”的效果。但注意，这会导致无限循环重试，生产环境必须配最大重试次数或死信队列。

3. 发布订阅与路由模式：从“点对点”到“一对多分发”

前面两种模式再怎么折腾，本质上还是“一条消息只有一个消费者处理”。从发布订阅开始，画风突变：生产者一条消息，多个消费者要同时收到。典型场景就是商品库存变更后，通知订单服务、缓存服务、搜索引擎各来一份。

3.1 发布订阅模式：所有人都得知道这件事

发布订阅的核心交换机类型是 fanout。fanout 的规则就是广播：不管 routing key 写什么，只要队列绑定到了这个交换机，消息就复制一份送过去。

生产者（emit_log.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', exchange_type='fanout') message = ' '.join(sys.argv[1:]) or "info: log message" channel.basic_publish(exchange='logs', routing_key='', body=message.encode()) print("广播消息:", message) connection.close()

消费者（receive_logs.py）：

import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', exchange_type='fanout') result = channel.queue_declare(queue='', exclusive=True) queue_name = result.method.queue channel.queue_bind(exchange='logs', queue=queue_name) def callback(ch, method, properties, body): print("收到广播:", body.decode()) channel.basic_consume(queue=queue_name, on_message_callback=callback) print("等待广播消息...") channel.start_consuming()

这里有几个细节要抠。第一，channel.exchange_declare(exchange='logs', exchange_type='fanout')，如果交换机已存在，重复声明只要类型一致就不会报错；但一旦类型不一致，声明时会直接抛异常。所以设计初期要定好交换机命名规范，别把“logs”这个名字既当 fanout 又当 direct 用。

第二，消费者里的queue_declare(queue='', exclusive=True)是让 RabbitMQ 自动生成一个随机队列名，并把队列设置为“排他”——消费者断开连接后队列自动删除。这就是发布订阅模式的精髓：每个消费者独享一个临时队列，都会收到完整消息副本。

注意：如果消费者是代码执行完就退出的一次性脚本，那消息发出来后没人监听，消息照样丢。RabbitMQ 的 fanout 交换机只管“投递时有哪些活着的绑定”，它不会像队列一样帮你缓存消息等消费者上线。所以发布订阅模式适合“消费者长期在线”的场景，比如多个微服务同步接收配置变更推送。

3.2 路由模式：只让关注“错误级别”的人收到

发布订阅的另一个极端是：所有人收到全量消息，会造成大量无效消费。比如日志系统里，写文件的服务其实只关心 error 级别的日志，没必要让它也接收一条 debug 日志。这时就得用 direct 交换机，把 routing key 当作“消息标签”，队列只接收固定标签的消息。

生产者（emit_log_direct.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', exchange_type='direct') severity = sys.argv[1] if len(sys.argv) > 1 else 'info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish( exchange='direct_logs', routing_key=severity, body=message.encode() ) print("发送 [%s] 消息: %s" % (severity, message)) connection.close()

消费者（receive_logs_direct.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', exchange_type='direct') severities = sys.argv[1:] if not severities: sys.exit("用法: receive_logs_direct.py [info] [error] [warning]") result = channel.queue_declare('', exclusive=True) queue_name = result.method.queue for severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity) def callback(ch, method, properties, body): print("收到 [%s] 消息: %s" % (method.routing_key, body.decode())) channel.basic_consume(queue=queue_name, on_message_callback=callback) print("等待消息...") channel.start_consuming()

这段代码的难点在于“一个队列绑定多个 routing key”——上面这个消费者同时绑定了info、error、warning三个键，就能收到三种级别的日志；而另一个消费者只绑定了error一个键，就只会收到错误日志。你有没有发现，这种按标签精确过滤的模型，本质就是“可控的定向推送”。

在实际业务里，direct 模式最常见的使用方式是“订单状态通知”。假设有一个order_status交换机，某个队列绑定routing_key='order.paid'，另一个队列绑定routing_key='order.cancelled'。支付服务只发order.paid，物流服务只监听order.paid，库存回滚服务只监听order.cancelled。各个系统互不干扰。

避坑提示：direct 交换机下，routing_key匹配是“完全相等”匹配，不能带通配符。如果你发现自己需要匹配order.*之类带星号的规则，说明该换 topic 交换机了，别硬在 direct 上搞字符串拼接代替通配符。

4. 主题模式与RPC模式：高级玩法与跨系统同步

前三对模式大家可能都见过，主题模式和RPC模式就更容易让人犯迷糊，因为牵扯的概念更多。先说主题模式：它是路由模式的“模糊匹配升级版”，靠*（匹配一个单词）和#（匹配零个或多个单词）两个通配符，可以表达很多复杂的业务筛选规则。而RPC模式则是在消息队列上实现“同步调用”，是很多支付系统、订单系统在做跨服务调用时喜欢采用的姿势。

4.1 主题模式：用通配符做消息的复杂筛选

假设你在做大宗商品的行情推送系统。不同客户订阅不同品种、不同交易所的行情。如果用 direct 模式，得给每个订阅组合建一条队列，非常离谱。主题模式用一条匹配规则就搞定。

生产者（emit_log_topic.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', exchange_type='topic') routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish( exchange='topic_logs', routing_key=routing_key, body=message.encode() ) print("发送 [%s] 消息: %s" % (routing_key, message)) connection.close()

消费者（receive_logs_topic.py）：

import pika, sys connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', exchange_type='topic') binding_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' result = channel.queue_declare('', exclusive=True) queue_name = result.method.queue channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key) def callback(ch, method, properties, body): print("收到 [%s] 消息: %s" % (method.routing_key, body.decode())) channel.basic_consume(queue=queue_name, on_message_callback=callback) print("等待消息...") channel.start_consuming()

这样，如果你启动消费者时传入"*.stock"，就能收到所有以.stock结尾的行情消息，比如shenzhen.stock、shanghai.stock；如果你传入"#.gold"，就能匹配到任意层级前缀、最后一段是黄金品种的消息。在生产环境里，主题模式非常适合做基于标签的推送中心、数据分发管道，甚至内部告警系统。

主题模式的匹配规则有三个地方容易让新手卡壳：

• *必须匹配一个完整的单词（点号分隔的部分），不能匹配跨级内容。例如*.stock能匹配sh.stock，但不能匹配cn.sh.stock。
• #可以匹配零个或多个单词，所以#.stock可以同时匹配stock和cn.sh.stock。
• 如果一个消息的 routing key 没有任何队列能匹配，消息会被直接丢弃（或者进死信交换机）。

所以设计 routing key 的命名规范极其重要。我之前负责的一个项目里，各业务方乱起键名，一会儿用order.paid，一会儿用orderPaid，最后 topic 匹配规则怎么都写不准，光排查就花了一周。建议在一开始就定死“领域.动作.版本”之类的三段式规范，比如trade.order.v1，后面接新业务时会省心很多。

4.2 RPC模式：用消息队列做同步调用

很多人以为消息队列一定是异步的，其实 RabbitMQ 完全可以实现同步请求-响应——客户端发消息后阻塞等待，服务端处理完把结果写到另一个队列里，客户端再取出。这个模式在微服务架构里常用于跨语言调用：比如你用 Python 写一个计算服务，被 Java 的上游系统调用。

RPC客户端（rpc_client.py）：

import pika, uuid class FibonacciRpcClient(object): def __init__(self): self.connection = pika.BlockingConnection( pika.ConnectionParameters(host='localhost') ) self.channel = self.connection.channel() result = self.channel.queue_declare(queue='', exclusive=True) self.callback_queue = result.method.queue self.channel.basic_consume( queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True ) def on_response(self, ch, method, properties, body): if self.corr_id == properties.correlation_id: self.response = body def call(self, n): self.response = None self.corr_id = str(uuid.uuid4()) self.channel.basic_publish( exchange='', routing_key='rpc_queue', properties=pika.BasicProperties( reply_to=self.callback_queue, correlation_id=self.corr_id ), body=str(n).encode() ) while self.response is None: self.connection.process_data_events() return int(self.response) rpc = FibonacciRpcClient() print("请求计算 fib(30)") response = rpc.call(30) print("结果:", response)

RPC服务端（rpc_server.py）：

import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost')) channel = connection.channel() channel.queue_declare(queue='rpc_queue', durable=True) def fib(n): if n in (0, 1): return n a, b = 0, 1 for _ in range(n-1): a, b = b, a + b return b def on_request(ch, method, properties, body): n = int(body) print("计算 fib(%s)" % n) response = fib(n) ch.basic_publish( exchange='', routing_key=properties.reply_to, properties=pika.BasicProperties( correlation_id=properties.correlation_id ), body=str(response).encode() ) ch.basic_ack(delivery_tag=method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(queue='rpc_queue', on_message_callback=on_request) print("等待RPC请求...") channel.start_consuming()

这里面最重要的设计是correlation_id——客户端在发出请求时生成一个唯一ID，响应消息里必须带上同一个ID，客户端才认得这条响应是自己的。假设有两个客户端同时发来请求，服务端把结果返回给同一个回调队列，如果没有correlation_id，客户端之间就乱套了。这个机制跟 HTTP 请求的 traceId、请求ID是一个思路，本质是“给消息打标”。

实际生产中使用 RPC 模式时，我做几个额外提醒：

• 上面代码里客户端是“阻塞等待单条消息”，性能很低。真实场景建议用异步回调 + 超时控制，例如给每个请求设置3秒超时，超过就丢到重试队列。
• reply_to队列建议用临时队列，服务端处理完直接响应；别把回调队列做成持久化，否则客户端一多，积压的响应消息会把人搞疯。
• RPC 模式固然方便，但不要把消息队列当成唯一的跨系统调用通道。能异步就异步，能 HTTP 就 HTTP，RPC模式下消息队列反而容易变成性能瓶颈。

4.3 高级模式的坑：临时队列、correlation_id和消息超时

写 RPC 与 topic 模式最容易犯的错误有三个：一是回调队列声明为持久化队列，导致回调消息永久堆积；二是忘记给 RPC 请求设超时，客户端一旦消息走丢就永远阻塞；三是把reply_to和correlation_id写反——两个都丢了的话根本没法定位响应归属。前两个我之前都踩过。

5. 六大模式选型与架构决策：什么场景配什么模式

讲完六种模式之后，大部分人的下一步问题是：脑子里一堆概念，但真要动手设计系统时怎么选？我结合几个真实项目场景，把选型逻辑做一个系统性梳理。你可以把它当作一张速查表来用。

5.1 模式适用场景对照表

这张表里最重要的判别维度有两个：一是“这条消息是要被所有消费者消费一遍，还是只给其中一个消费”，如果是前者直接考虑 fanout；二是“消费者对消息的筛选规则是精确还是模糊”，模糊就上 topic，精确小场景用 direct 就行。

5.2 用一个完整案例串起模式选型：电商订单履约系统

我们拿一个电商系统做例子，把六种模式串起来讲：

• 用户下单后，支付系统收到支付成功回调：此时需要用工作队列把订单消息发给多个异步任务——更新订单状态、扣减库存、发送短信通知。这里如果用发布订阅会导致三个任务各自重复消费全部消息，浪费资源；用工作队列则每条消息只被一个消费者处理一次。
• 假设订单状态变更后，有三个下游系统都想知道，分别是履约系统、数据分析系统、用户积分系统。这三个系统都必须收到同一份“订单已支付”的消息，这时候换成发布订阅（fanout）最合理。
• 但如果只是给“订单催付提醒”场景发消息，只有用户服务需要消费，其他服务不关心，就用路由模式（direct），routing_key 设为order.remind，只有订阅了这个 tag 的队列能收到。
• 如果业务复杂到营销系统想按活动类型、商品类目、渠道来源等多维标签订阅订单消息，主题模式（topic）才够用。假设每个订单消息的 routing key 设计成order.sku.category.channel，营销活动队列绑定order.*.promotion.*，就能精准接收大促相关的订单。

从这里你能看出一个关键点：模式不是孤立选的，而是“围绕同一个订单事件，在不同链路用不同模式”。这也是 RabbitMQ 的魅力——看起来只有个队列，实际是一套可编排的消息路由系统。

5.3 选型时的三条血泪经验

第一，不要为了模式而模式。新手容易觉得“我这系统用了 topic 模式才高级”，业务简单的场景里 topic 只会增加排查复杂度。先问自己：现有业务真的需要模糊匹配吗？不需要就老实回 direct。

第二，RabbitMQ 模式下消息有序性问题必须提前设计。工作队列启用多个消费者后，消息的消费顺序天然不可控。对顺序敏感的场景（如订单状态机流转），要么只在单消费者队列上操作，要么用“同一业务的 routing key 固定到同一队列+单消费者”。

第三，交换机、队列、绑定关系的命名要像数据库表设计一样有规范。线上环境最怕的就是谁知道哪个交换机负责什么。我一般坚持“交换机名 = 业务域.消息主题”，队列名 = “业务域.消费者组.用途”，绑定关系用配置文件统一管理，禁止临时在代码里创建一个新队列。

6. 常见问题与排查技巧实录：从连接断开到消息堆积

最后聊一聊实战里大家最常被坑的几个地方。作为一个被 RabbitMQ 折磨过很多次的人，我把自己积累的排查经验整理成了一份问题速查表，希望能帮你少走弯路。这些问题不分模式，六种模式下的运维姿势都一样。

6.1 消息丢失与重复消费

消息丢失通常出现在三个环节：生产者发送时、RabbitMQ 存储时、消费者处理时。

• 生产者发送失败：默认情况下basic_publish不会确认消息是否到达交换机。要拿到确认结果，需要开启发布者确认模式（channel.confirm_delivery()或异步 ConfirmListener），如果失败就重发。
• RabbitMQ 存储时丢失：只有同时满足“队列持久化 + 消息持久化”才安全。更稳妥的方案是为重要交换机配一个备用交换机（Alternate Exchange），把无法路由的消息转到死信队列再告警。
• 消费者处理时丢失：核心就在于手动 ack 的时机。之前说过的basic_ack一定要在业务处理完成后再调。

重复消费几乎是难以完全避免的，只要出现网络重试，就可能重复。所以业务处理逻辑必须支持幂等：用消息里的唯一业务ID建唯一索引，消费前先查是否已处理过；或者在处理逻辑里用分布式锁挡住并发重复。

6.2 连接断开、心跳超时与消息堆积

RabbitMQ 默认心跳超时是60秒，如果你的消费者处理一条消息超过60秒，中间没有给 Broker 发送任何心跳，连接会被判定为死连接，导致消息被重新投递给其他消费者，进而引发重复消费。解决思路两条：处理耗时长的任务改用异步线程池 + 快速 ack，然后用单独的写入线程去落库；或者调大heartbeat参数，但这不是根治方法。

消息堆积的原因就比较多了：消费能力不足、消费者宕机、prefetch设置不合理、bind 关系写错导致消息进不了队列等等。我的排查顺序是：

1. 先看队列积压数量：rabbitmqctl list_queues name messages consumers
2. 看消费者是否存活：rabbitmqctl list_consumers
3. 看消息是否一直 unacked：如果unacked高，多半是消费者处理不过来或卡死了
4. 看有没有消息进了死信队列：rabbitmqctl list_queues name messages messages_unacknowledged

6.3 一则基于真实经验的排查实战：日志堆积事故

某次我们内部日志系统的消费者进程没有崩溃，但队列积压到了几十万条。排查发现，日志消费者里调用了外部 HTTP 接口，接口超时时间设成了30秒，而 HTTP 连接池又没有设置最大等待数，导致所有消费者线程全部卡在等待响应上。表面上消费者进程还活着，但实际上一个消息都处理不了，unacked 不断增加，RabbitMQ 则不断给消费者推送新消息，最终堆积如山。

后面我们做的调整是：消费者只负责把日志写入本地文件或直接入库，外部接口的调用放到另一个独立的异步处理队列，用专门的 worker 去调用，并设置超时和熔断。这也是我给所有处理耗时任务的团队的统一建议——不要在消费者里做耗时外部调用，除非有完整的超时、重试、熔断机制。

6.4 死信队列和延迟消息的正确打开方式

很多业务场景需要延迟处理，比如订单15分钟未支付自动关单、用户注册后30分钟发优惠券。RabbitMQ 本身没有原生的延迟消息，但可以通过“死信交换机 + 延迟队列”实现：声明一个普通队列，队列设置x-message-ttl，比如30秒，再给这个队列指定x-dead-letter-exchange指向正式处理队列的交换机，消息30秒无人消费就会自动转到死信交换机，从而被“延迟投递”到真正消费者。

这个方案是小成本延迟消息的标准姿势，不需要额外插件。唯一要注意的是 RabbitMQ 的死信转发是基于“队列TTL到期”的扫描机制，可能会有一小段时间误差，对秒级精度的延迟场景不合适；且如果前面的队列堆积了很多消息，后续消息的延迟时间会不稳定。如果对时间精度要求高，建议直接装延迟插件或选用专门的延迟队列中间件。

我个人在实际操作中最常用的组合是“工作队列处理异步任务 + 主题模式做数据分发 + 死信队列兜底异常”，这套组合基本能覆盖大部分业务需求。读代码的人只要把“交换机决定分发策略，队列决定存储归属，绑定关系决定谁消费”这三句话记住，六种模式再怎么变你都能一眼看穿。RabbitMQ 不难，难的是在每次做技术选型的时候，多问自己一句“这个场景真的需要这么复杂的路由规则吗”，答案往往写在业务复杂度里。