RabbitMQ 消费确认与预取manual ack、reject、nack 与 QoS 的取舍1. 先看一个真实故障消息去哪了假设你负责一个订单履约服务。上游把「订单已支付」事件写进 RabbitMQ下游消费者负责扣减库存、发短信、写履约单。上线第一周一切正常第二周某个凌晨消费者进程被 OOM Killer 杀掉重启后运维发现有 300 多笔订单明明支付成功了但库存没扣、短信没发。翻日志消费者在处理到第 120 条消息时被打断后面 180 条根本没被消费。问题出在哪里RabbitMQ 默认使用自动确认autoAck。消费者从队列拿到消息的一瞬间Broker 就把这条消息删掉了不管你后续业务有没有跑完。消费者进程崩溃那些「已投递但未处理完」的消息就永久消失了。所以自动确认的语义是「投递即成功」而不是「处理即成功」。要修复它你需要一套手动确认manual ack机制业务处理成功后才告诉 Broker「这条可以删了」处理失败或进程挂掉时让消息回到队列重新投递。但一旦引入手动确认新问题马上出现如果一次拉 100 条、并发处理某条失败后要重投会不会把已经处理成功的也一起重投如果某条消息永远处理失败会不会无限循环、把 CPU 打满消费者卡死但连接不断Broker 会不会一直等它这些正是本文要回答的问题。2. 一句话模型与全局框架先把核心模型用一句普通中文说清楚RabbitMQ 的消费确认本质上是消费者与 Broker 之间围绕「这条消息我负责到底了没有」的一次显式对话而预取QoS决定的是这条对话一次可以并行进行多少条。为了不迷失在 API 细节里先把它拆成三个角色和一条时间线生产者Producer把消息发到交换机。交换机与队列Exchange / Queue交换机按路由规则把消息放进队列队列是消息真正排队等待被取走的地方。消费者Consumer从队列取消息、执行业务、回执确认。三者与 Broker 之间一次完整流转的顺序如下生产者 Broker(交换机队列) 消费者 | 1. publish | | |-----------------------| | | | 2. 入队, 等待投递 | | | 3. deliver (带 deliveryTag)| | |---------------------------| | | | 4. 执行本地业务 | | | 5. 回执 ack/nack/reject | |---------------------------| | | 6a. ack - 删除消息 | | | 6b. nack - 重新入队/死信 |这张图里有两个关键点后文会反复用到。第一deliveryTag是 Broker 在单个信道Channel上给每条投递消息编的序号从 1 开始递增确认时必须带着这个序号且确认只在该信道内有效。第二ack 与 nack/reject 的区别不是「成功与失败」的情绪差别而是Broker 后续怎么处置这条消息是删除还是放回队列还是丢进死信交换机。理解了这个框架后面所有 API 都是在回答两个问题这次回执针对哪条或哪批消息Broker 收到后做什么3. AMQP 里的确认到底确认了什么很多人以为 ack 是「客户端告诉服务端我收到了」。在 RabbitMQ 的语境里更准确的说法是消费者告诉 Broker「这条消息的归属权可以从我身上摘掉了你可以按你的策略处置它」。这句话有两层含义。第一层是归属权转移。消息一旦被投递给某个消费者但未确认Broker 会把它标记为 unacked。此时消息不在队列头部等待投递也不算已删除而是「押」在这个消费者对应的信道账上。如果消费者断开连接或信道关闭Broker 会把该信道上所有 unacked 消息重新入队投给其他消费者或自己重连后再取。这是可靠性的根本保障崩溃不会丢消息代价是可能重复。第二层是批量语义。Broker 不要求你逐条确认。basicAck(deliveryTag, multipletrue)表示「这个 tag 以及它之前所有未确认的都算完成」。批量确认能显著减少网络往返提升吞吐但代价是精度下降如果你只想确认第 100 条却用了 multipletrue那 1 到 99 条也会被一并确认一旦 1 到 99 里有处理失败的就再也没机会重投了。这里最容易误解的是ack 的「成功」完全由你的业务代码定义RabbitMQ 无法校验。你可以在根本没写数据库的情况下调用 ack也可以在写库成功后忘记 ack。RabbitMQ 只记录「你有没有回执」不记录「你回执得对不对」。因此确认逻辑必须和业务事务的边界对齐这是后面幂等设计的伏笔。4. 手动确认模式把删除权从 Broker 拿回来手动确认的开关是消费者注册时的autoAckfalse在 Java 客户端里是basicConsume的autoAck参数。这个开关一旦打开你就有义务在合适时机调用basicAck否则消息会一直挂在 unacked 里占内存、占队列配额最终触发流控。手动确认的典型正确姿势是「先业务、后确认」并且要用 try/catch 包住业务异常时走 nack 或 reject。下面这段是完整的可运行示例环境是 JDK 17 com.rabbitmq:amqp-client:5.20.0Broker 是本机localhost:5672默认 guest/guest。它声明一个持久化队列消费端显式autoAckfalse业务成功才 ack。// ManualAckDemo.java —— 最小可运行的手动确认消费者importcom.rabbitmq.client.*;importjava.nio.charset.StandardCharsets;publicclassManualAckDemo{privatestaticfinalStringQUEUEdemo.order.paid;publicstaticvoidmain(String[]args)throwsException{ConnectionFactoryfactorynewConnectionFactory();factory.setHost(localhost);factory.setPort(5672);factory.setUsername(guest);factory.setPassword(guest);Connectionconnectionfactory.newConnection();Channelchannelconnection.createChannel();// 持久化队列durabletrue第三个参数 exclusivefalse自动删除falsechannel.queueDeclare(QUEUE,true,false,false,null);// 生产一条消息方便直接消费演示channel.basicPublish(,QUEUE,MessageProperties.PERSISTENT_TEXT_PLAIN,order-1001 paid.getBytes(StandardCharsets.UTF_8));// autoAckfalse 是手动确认的关键开关DeliverCallbackdeliverCallback(consumerTag,delivery)-{longtagdelivery.getEnvelope().getDeliveryTag();StringbodynewString(delivery.getBody(),StandardCharsets.UTF_8);try{// 模拟业务真实场景这里是一次数据库写 一次 RPChandleBusiness(body);// 业务成功后才确认multiplefalse 表示只确认这一条channel.basicAck(tag,false);System.out.println([ack] body);}catch(Exceptione){// 业务失败拒绝并重新入队交给其他消费者重试channel.basicNack(tag,false,true);System.err.println([nack-requeue] body causee.getMessage());}};// 注意第二个参数 autoAckfalsechannel.basicConsume(QUEUE,false,deliverCallback,consumerTag-{});System.out.println(consumer started, waiting...);Thread.sleep(10_000);channel.close();connection.close();}privatestaticvoidhandleBusiness(Stringbody){if(body.contains(poison)){thrownewIllegalStateException(业务处理失败);}// 正常业务写库、调用下游等}}运行方式启动本地 RabbitMQ 后javac -cp amqp-client-5.20.0.jar ManualAckDemo.java再java -cp .:amqp-client-5.20.0.jar:slf4j-api-1.7.36.jar ManualAckDemo。预期输出是先打印consumer started随后[ack] order-1001 paid。关键点有三处。第一basicAck放在handleBusiness之后而不是之前这就是「先业务后确认」。第二失败时用了basicNack(tag, false, true)第三个参数requeuetrue表示回队列——但这个选择有风险下一节会专门讲。第三multiplefalse意味着精确确认单条这是初学阶段最安全的默认选择。5. basicQos 预取一次别给我太多上一节的代码有一个隐藏问题Broker 会把队列里的消息尽可能快地推给消费者如果你的消费者只有一个线程处理或者处理速度远低于投递速度unacked 消息会快速堆积。放大到多消费者场景还会出现「忙的消费者手里堆满未确认消息闲的消费者没活干」的倾斜现象。这就是预取prefetch要解决的。basicQos(prefetchCount, prefetchSize, global)是消费者端给 Broker 的限制在同一时间最多允许有多少条「已投递但未确认」的消息挂在我这里。prefetchCount是最常用的参数prefetchSize一般填 0表示不按字节数限制global的语义要特别小心。// 伪代码展示 QoS 的两种作用域不能直接运行// 作用域一仅对之后新建的消费者生效RabbitMQ 常见语义channel.basicQos(10,false/* prefetchSize */,false/* global */);channel.basicConsume(QUEUE,false,deliverCallback,cancelCallback);// 作用域二对当前信道上的所有消费者生效globaltruechannel.basicQos(10,false,true);先记住一个最小模型QoS 不是限速器而是「在途额度」。它限制的是「未确认」的数量不是每秒投递数量。只要你不 ack额度就不会释放。因此prefetchCount1意味着严格的一条一确认、串行处理prefetchCount100意味着允许 100 条同时在途适合高并发批量处理。预取值吞吐特性内存与公平性典型场景1低几乎串行内存占用最小最公平处理慢、单条成本高的任务10~50中等较均衡普通业务消费多数场景的起点100~500高unacked 堆积明显消费者倾斜风险上升批量计算、写入吞吐优先0不限制最高但不可控容易 OOM 并触发流控极少数实验场景不建议生产注意一个反直觉的地方预取值越大单条消息的端到端延迟可能越高。因为消息被推到某个消费者后如果这个消费者在慢慢处理前面的消息后面的消息只能等着而另一个空闲消费者却取不到。所以「提高预取等于提高性能」只在批量、均匀、消费者数量少时成立。6. reject 与 nack拒绝之后的两种命运业务失败时你不能只 ack也不能放着不管。RabbitMQ 给了两个 APIbasicReject和basicNack。它们的关系可以用一句话概括reject 是 nack 的单条版本nack 能批量reject 不能。API能否批量参数常见用途basicAck(tag, multiple)能deliveryTag、multiple业务成功删除消息basicReject(tag, requeue)不能deliveryTag、requeue单条失败明确重投或丢弃basicNack(tag, multiple, requeue)能deliveryTag、multiple、requeue批量失败统一重投或丢弃requeuetrue表示把消息放回队列requeuefalse表示丢弃。丢弃并不等于消息消失得无影无踪——如果队列配置了死信交换机DLX消息会被转发到死信队列供后续人工排查或延迟重试。这里把 DLX 的角色说清楚它是队列的一个属性消息因 nackrequeuefalse、TTL 过期或队列满而被丢弃时Broker 会把它路由到 DLX。一个常见错误是把basicReject(tag, false)当成「重试」。它其实是丢弃。要重试必须requeuetrue但重试次数无法由 RabbitMQ 控制只能靠你自己记录。下面是把「有限重试 死信」串起来的核心片段属于简化代码需配合 DLX 声明才能运行// 简化代码通过 x-death 头判断重试次数超过阈值就丢进死信DeliverCallbackcallback(tag,delivery)-{longdTagdelivery.getEnvelope().getDeliveryTag();MapString,Objectheadersdelivery.getProperties().getHeaders();intretryextractRetryCount(headers);// 从 x-death 头里解析重试次数try{handleBusiness(newString(delivery.getBody(),StandardCharsets.UTF_8));tag.basicAck(dTag,false);}catch(Exceptione){if(retry3){tag.basicNack(dTag,false,true);// 前三次重投}else{tag.basicNack(dTag,false,false);// 超过阈值进死信队列}}};x-death是 Broker 在消息因死信原因被投递时写入的头部记录了死信原因和次数。它的具体字段在不同版本和不同死信原因下略有差异所以解析逻辑要按你实际 Broker 版本验证不要盲抄。如果你不想依赖它更稳妥的做法是在业务侧用一张「消息去重/重试计数表」自己维护。7. 重新入队风险看似安全的 requeuetruerequeuetrue看起来是最省心的失败处理失败了放回去再试嘛。但它藏着两个真实的生产事故源饥饿和风暴。先说饥饿。RabbitMQ 重新入队时默认会把消息放回队列头部附近。如果队列里只有这一条「毒消息」反复失败而你只有一个消费者它会被立刻再次投给同一个消费者导致队列后面所有正常消息被无限期卡住。这就是经典的「毒消息阻塞」。多消费者只能缓解不能根治如果每个消费者都立刻拿到它并失败整条队列的有效吞吐都会下降。再说风暴。当一批消息因为同一个下游故障比如数据库连接池耗尽集体失败requeuetrue会让它们几乎立刻回到队列并再次被投递形成高频重试循环。这不仅把 CPU 和网络打满还会持续刷日志掩盖真正的故障原因。下面这张时序把两种风险串起来毒消息场景(单消费者, requeuetrue): 消费者: 取 msgb - 失败 - nack(requeuetrue) - 重新入队(靠前) 消费者: 又取到 msgb - 失败 - nack(requeuetrue) - 循环... 队列: [b][1][2][3] - 1/2/3 永远轮不到 批量故障场景(下游数据库不可用): msg1..msg100 全部失败 - 全部 requeue - 立刻重新投递 - 再次全部失败 时间轴: 重试间隔约等于 0, 直到数据库恢复或消息被拖入死信正确的工程做法是给重试加上时间和次数约束。时间约束靠延迟队列或 TTL DLX次数约束靠你自己记录的重试计数。当重试超限后用basicNack(tag, false, false)把消息送到死信队列让主队列继续前进。这个组合的完整链路是主队列 - 处理失败且未超限 - 重投 - 超限 - DLX - 死信队列 - 人工或定时补偿。8. 消费者超时与长时间未确认还有一个隐蔽的失败模式消费者既没 ack 也没 nack业务卡在那里。常见原因是下游 RPC 没有超时、数据库慢查询、死锁或线程被阻塞。此时消息一直处于 unacked 状态占用预取额度消费者看起来「活着」但实际不干活。RabbitMQ 有一个连接层面的consumer_timeout消费者确认超时默认 30 分钟可在配置中调整。一旦某条投递在该时间内没有确认Broker 会主动关闭这个信道unacked 消息全部重新入队。这个机制保护的是集群资源不被卡死但它的副作用是如果配置为 30 分钟而你的业务正常处理需要 31 分钟你的消息会被反复重投形成难以排查的重复消费。因此业务处理时间必须明显小于consumer_timeout或者你需要为长任务设计另一种模式先快速 ack 一条「已接收」记录再做异步处理用业务侧台账保证最终一致。很多团队用「先 ack 后处理」来绕过超时但这就把可靠性责任完全转移到了自己身上必须配套幂等和补偿机制否则会丢消息。9. 从一次投递到确认完整链路推演现在让一次真实消费完整走一遍把前面所有概念串起来。场景订单履约消费者预取 20业务是扣库存 发短信下游偶尔超时。[1] Broker 投递 msg(tag7) - 消费者, unacked7 [2] 消费者线程从处理池取线程执行 [3] 扣库存成功 [4] 发短信超时, 抛异常 [5] 判断重试次数 retry0 3 [6] basicNack(7, false, true) - Broker 重新入队 [7] Broker 再次投递 msg(tag9) - 可能给同一或另一消费者 [8] 再次失败... 直到 retry3 [9] basicNack(tag, false, false) - 消息进死信队列 [10] 死信消费者记录工单, 人工介入这个链路里有三个设计决策点。第一在第 6 步和第 9 步之间必须有重试计数否则永远出不去。第二第 7 步的重新投递可能换消费者所以业务必须幂等——同一条消息被处理两次扣库存不能扣两次。第三第 10 步的死信不是终点要有可观测的告警和补偿流程否则死信队列会变成黑洞。下面给出第二个完整示例带幂等键和重试计数的消费者。环境同上额外建议你在 MySQL 里有一张t_msg_log(msg_id, status)表。这里用内存 Map 模拟幂等表保证单进程可运行生产请替换为数据库或 Redis。// IdempotentConsumer.java —— 幂等 有限重试的完整消费者importcom.rabbitmq.client.*;importjava.nio.charset.StandardCharsets;importjava.util.Map;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.atomic.AtomicInteger;publicclassIdempotentConsumer{privatestaticfinalStringQUEUEdemo.order.paid.idem;privatestaticfinalintMAX_RETRY3;// 模拟幂等表生产环境请换成数据库唯一索引或 Redis SETNXprivatestaticfinalMapString,BooleanDONEnewConcurrentHashMap();// 模拟重试计数生产环境请持久化否则重启后计数丢失privatestaticfinalMapString,AtomicIntegerRETRYnewConcurrentHashMap();publicstaticvoidmain(String[]args)throwsException{ConnectionFactoryfnewConnectionFactory();f.setHost(localhost);f.setPort(5672);f.setUsername(guest);f.setPassword(guest);try(Connectionconnf.newConnection();Channelchconn.createChannel()){ch.queueDeclare(QUEUE,true,false,false,null);ch.basicQos(20);// 预取 20控制 unacked 上限DeliverCallbackonDeliver(consumerTag,delivery)-{longdTagdelivery.getEnvelope().getDeliveryTag();StringmsgIddelivery.getProperties().getMessageId();StringbodynewString(delivery.getBody(),StandardCharsets.UTF_8);if(msgIdnull){// 没有幂等键的消息直接拒绝避免无法去重ch.basicNack(dTag,false,false);return;}if(DONE.containsKey(msgId)){// 幂等命中直接确认ch.basicAck(dTag,false);return;}try{handleBusiness(body,msgId);DONE.put(msgId,true);ch.basicAck(dTag,false);}catch(Exceptione){inttimesRETRY.computeIfAbsent(msgId,k-newAtomicInteger()).incrementAndGet();booleanrequeuetimesMAX_RETRY;ch.basicNack(dTag,false,requeue);// 未超限重投超限进死信System.err.println(fail msgIdmsgId timestimes requeuerequeue);}};ch.basicConsume(QUEUE,false,onDeliver,t-{});System.out.println(idempotent consumer started, prefetch20);Thread.sleep(15_000);}}privatestaticvoidhandleBusiness(Stringbody,StringmsgId){// 真实业务扣库存 发短信必须使用 msgId 保证下游可去重System.out.println(handling msgId bodybody);}}预期行为同一条消息即使因 nack 被重投第二次到达时msgId已在DONE里直接 ack不会重复扣库存。失败消息会在重试 3 次后走死信。最容易改错的地方有两个msgId必须在生产端就设置好并全局唯一重试计数如果只放内存进程重启后会归零导致消息无限重试生产必须持久化或改用x-death头。10. QoS、ack 与业务并发怎么配合当消费者内部用线程池并发处理时QoS 的语义会变得微妙。假设预取是 20消费者用 4 个线程处理那么同一时刻最多 20 条在途、最多 4 条在跑其余 16 条在内存队列里排队。这没问题但要注意两点。第一ack 必须在处理线程里调用且要小心 Channel 的线程安全。RabbitMQ Java 客户端的Channel不是线程安全的多线程同时 ack 需要自己加锁或者为每个线程开独立 Channel。很多「偶发 ack 无效」的 bug 都源于此。共享 Channel 上的 deliveryTag 序号是全局递增的批量 ack 时尤其容易误伤。第二预取要略大于并发线程数但不能太大。如果预取等于线程数当某条消息处理特别慢时其他空闲线程就没有新消息可拿吞吐下降如果预取远大于线程数内存里堆积的业务数据会变多且消费者倾斜更严重。一个可用的经验起点是「预取 并发线程数 × 2」再根据压测调整。第三种完整场景是生产端配合消息带messageId队列声明 DLX消费端使用预取与有限重试。下面的命令块展示如何用rabbitmqadmin或管理 API 观察 unacked 与死信情况这属于可复现的排查操作前置条件是开启 management 插件。# 查看队列深度、消费者数、未确认数需要 management 插件与本地 admin 工具rabbitmqadmin list queues name messages messages_unacknowledged consumers# 输出示例真实字段随版本略有差异:# --------------------------------------------------------------------# | name | messages | messages_unacknowledged | consumers |# --------------------------------------------------------------------# | demo.order.paid.idem | 120 | 20 | 1 |# --------------------------------------------------------------------# 查看某个队列的死信配置与消息数rabbitmqadmin list queues name arguments dead_letter_exchangemessages_unacknowledged长期等于预取值通常说明消费者的处理速度跟不上投递速度或者有消息卡住没确认messages持续增长则说明整体消费能力不足。这两个指标是排查消费问题的第一入口。11. 设计取舍可靠、吞吐与重复之间的三角到这里可以把所有取舍收拢成一张决策表。核心矛盾是你要可靠性就要手动确认和重试你要重试就可能重复消费你要高吞吐就要放大预取但放大预取会让失败重投的粒度变粗、内存压力变大。目标推荐配置需要接受的代价宁可重复、不可丢失manual ack 有限重试 死信 幂等业务侧必须实现幂等重复不可避免高吞吐批量处理prefetch 100~500 批量 ack(multipletrue)精度下降失败时可能整批重投严格顺序、低延迟prefetch1 单消费者吞吐低扩展性差长任务先 ack 再异步处理 本地任务表可靠性责任转移到业务侧失败隔离nack(requeuefalse) DLX需要维护死信消费与告警有一个反直觉但重要的结论在分布式系统里「恰好一次」几乎不可能工程上追求的是「至少一次 幂等」。所以与其纠结怎么防止重复不如把幂等做扎实。幂等键可以是业务单号、消息messageId或者由「业务主键 操作类型」拼出来。另一个取舍是批量 ack 的边界。批量 ack 能省网络往返但它把多条消息的命运绑在一起。只有在「同一批消息要么都成功、要么都可安全重试」时才用且必须保证批内不会混入处理结果不同的消息。如果做不到就老老实实单条 ack。12. 生产实践建议与常见误区生产实践建议部分先给几条可以直接落地的规则。第一所有重要业务队列都关闭自动确认使用 manual ack。第二预取值从 20 起步压测后再调不要一上来就设 0。第三失败重试必须有次数上限和时间间隔推荐用 DLX TTL 做延迟重试而不是立刻requeuetrue。第四所有可能重复消费的业务都必须有幂等键和唯一约束。第五为每个消费队列配置死信队列并对死信数量做告警死信为 0 不代表正常突然增长才是信号。常见误区则集中在下面几点误区一以为autoAcktrue只是「自动帮我 ack 一下」。实际上它让消息在投递瞬间就被删除业务失败即丢失。误区二以为basicReject(tag, false)是重试。它是丢弃要重试必须requeuetrue。误区三以为requeuetrue一定会回到队尾。它通常会被放到靠近头部的位置从而阻塞后面的消息。误区四以为 QoS 是限速。它限制的是未确认数量不限制投递速率。误区五在业务成功前就 ack以换取「不重复」。这等于用丢消息换去重非常危险。误区六多个线程共享一个 Channel 并各自 ack。Channel 非线程安全可能引发连接层异常或确认错乱。13. 排障清单从现象到动作现象可能原因排查动作消息莫名消失使用了 autoAck 且业务失败检查消费者autoAck参数改为 falseunacked 长期不降业务卡住、未 ack、超时看messages_unacknowledged检查下游超时与线程栈队列积压但消费者空闲预取过小或消费者倾斜调大预取增加消费者或优化处理速度同一条消息反复重投毒消息 requeuetrue加最大重试次数超限进死信消息被重复消费重投 无幂等引入幂等键与唯一约束连接被 Broker 关闭消费者确认超时检查处理耗时与consumer_timeout配置排查顺序建议是先看队列指标深度、unacked、消费者数再看消费者日志是否 ack、是否异常循环最后看 Broker 日志是否有连接或信道被关闭。不要一上来就改预取值先把「有没有确认」这个根因定位清楚。14. 面试/复盘问题autoAcktrue和autoAckfalse在消息生命周期上的差别是什么分别在什么时机删除消息basicAck的multipletrue有什么语义什么场景下会误伤已经成功的消息basicReject和basicNack的区别是什么什么时候必须用 nackrequeuetrue会带来哪些风险如何设计有限重试basicQos限制的是什么为什么prefetch0通常不建议用于生产消费者超时触发后RabbitMQ 会做什么这对业务处理时间有什么要求为什么说「至少一次 幂等」比追求「恰好一次」更现实你会怎么设计幂等键这些问题可以当作团队评审的检查项如果你的消费者代码答不上第 2、4、7 题说明它在失败路径上的行为是不确定的。15. 总结把全文收回到一张框架图生产端负责让消息可路由、可去重Broker 负责排队、投递和在未确认时保留消息消费端负责在业务成功后确认、失败时受控重试、超限进死信。手动确认解决「丢消息」预取解决「堆积与倾斜」nack/reject 解决「失败怎么处置」死信与幂等解决「重试之后怎么办」。最后给一条可以直接执行的判断规则先关闭 autoAck、把预取设为 20、给失败加最大重试次数、给业务加幂等键、给队列加死信。这五步做完你就已经避开了绝大多数消费端事故。剩下的调优——预取调多大、并发开多少、批量 ack 用不用——都可以在压测中慢慢找答案。16. 参考资料RabbitMQ 官方文档Consumer Acknowledgements and Publisher ConfirmsRabbitMQ 官方文档Confirms (Publisher Confirms) 与 Consumer PrefetchRabbitMQ 官方文档Dead Lettering、TTL 与 Queue Length LimitRabbitMQ 官方文档Consumer Timeoutconsumer_timeout配置说明RabbitMQ Java Client API 文档Channel、DeliverCallback、basicAck、basicNack、basicQosAMQP 0-9-1 规范amqp.org《RabbitMQ 实战指南》相关章节朱忠华
RabbitMQ 消费确认与预取：manual ack、reject、nack 与 QoS 的取舍
RabbitMQ 消费确认与预取manual ack、reject、nack 与 QoS 的取舍1. 先看一个真实故障消息去哪了假设你负责一个订单履约服务。上游把「订单已支付」事件写进 RabbitMQ下游消费者负责扣减库存、发短信、写履约单。上线第一周一切正常第二周某个凌晨消费者进程被 OOM Killer 杀掉重启后运维发现有 300 多笔订单明明支付成功了但库存没扣、短信没发。翻日志消费者在处理到第 120 条消息时被打断后面 180 条根本没被消费。问题出在哪里RabbitMQ 默认使用自动确认autoAck。消费者从队列拿到消息的一瞬间Broker 就把这条消息删掉了不管你后续业务有没有跑完。消费者进程崩溃那些「已投递但未处理完」的消息就永久消失了。所以自动确认的语义是「投递即成功」而不是「处理即成功」。要修复它你需要一套手动确认manual ack机制业务处理成功后才告诉 Broker「这条可以删了」处理失败或进程挂掉时让消息回到队列重新投递。但一旦引入手动确认新问题马上出现如果一次拉 100 条、并发处理某条失败后要重投会不会把已经处理成功的也一起重投如果某条消息永远处理失败会不会无限循环、把 CPU 打满消费者卡死但连接不断Broker 会不会一直等它这些正是本文要回答的问题。2. 一句话模型与全局框架先把核心模型用一句普通中文说清楚RabbitMQ 的消费确认本质上是消费者与 Broker 之间围绕「这条消息我负责到底了没有」的一次显式对话而预取QoS决定的是这条对话一次可以并行进行多少条。为了不迷失在 API 细节里先把它拆成三个角色和一条时间线生产者Producer把消息发到交换机。交换机与队列Exchange / Queue交换机按路由规则把消息放进队列队列是消息真正排队等待被取走的地方。消费者Consumer从队列取消息、执行业务、回执确认。三者与 Broker 之间一次完整流转的顺序如下生产者 Broker(交换机队列) 消费者 | 1. publish | | |-----------------------| | | | 2. 入队, 等待投递 | | | 3. deliver (带 deliveryTag)| | |---------------------------| | | | 4. 执行本地业务 | | | 5. 回执 ack/nack/reject | |---------------------------| | | 6a. ack - 删除消息 | | | 6b. nack - 重新入队/死信 |这张图里有两个关键点后文会反复用到。第一deliveryTag是 Broker 在单个信道Channel上给每条投递消息编的序号从 1 开始递增确认时必须带着这个序号且确认只在该信道内有效。第二ack 与 nack/reject 的区别不是「成功与失败」的情绪差别而是Broker 后续怎么处置这条消息是删除还是放回队列还是丢进死信交换机。理解了这个框架后面所有 API 都是在回答两个问题这次回执针对哪条或哪批消息Broker 收到后做什么3. AMQP 里的确认到底确认了什么很多人以为 ack 是「客户端告诉服务端我收到了」。在 RabbitMQ 的语境里更准确的说法是消费者告诉 Broker「这条消息的归属权可以从我身上摘掉了你可以按你的策略处置它」。这句话有两层含义。第一层是归属权转移。消息一旦被投递给某个消费者但未确认Broker 会把它标记为 unacked。此时消息不在队列头部等待投递也不算已删除而是「押」在这个消费者对应的信道账上。如果消费者断开连接或信道关闭Broker 会把该信道上所有 unacked 消息重新入队投给其他消费者或自己重连后再取。这是可靠性的根本保障崩溃不会丢消息代价是可能重复。第二层是批量语义。Broker 不要求你逐条确认。basicAck(deliveryTag, multipletrue)表示「这个 tag 以及它之前所有未确认的都算完成」。批量确认能显著减少网络往返提升吞吐但代价是精度下降如果你只想确认第 100 条却用了 multipletrue那 1 到 99 条也会被一并确认一旦 1 到 99 里有处理失败的就再也没机会重投了。这里最容易误解的是ack 的「成功」完全由你的业务代码定义RabbitMQ 无法校验。你可以在根本没写数据库的情况下调用 ack也可以在写库成功后忘记 ack。RabbitMQ 只记录「你有没有回执」不记录「你回执得对不对」。因此确认逻辑必须和业务事务的边界对齐这是后面幂等设计的伏笔。4. 手动确认模式把删除权从 Broker 拿回来手动确认的开关是消费者注册时的autoAckfalse在 Java 客户端里是basicConsume的autoAck参数。这个开关一旦打开你就有义务在合适时机调用basicAck否则消息会一直挂在 unacked 里占内存、占队列配额最终触发流控。手动确认的典型正确姿势是「先业务、后确认」并且要用 try/catch 包住业务异常时走 nack 或 reject。下面这段是完整的可运行示例环境是 JDK 17 com.rabbitmq:amqp-client:5.20.0Broker 是本机localhost:5672默认 guest/guest。它声明一个持久化队列消费端显式autoAckfalse业务成功才 ack。// ManualAckDemo.java —— 最小可运行的手动确认消费者importcom.rabbitmq.client.*;importjava.nio.charset.StandardCharsets;publicclassManualAckDemo{privatestaticfinalStringQUEUEdemo.order.paid;publicstaticvoidmain(String[]args)throwsException{ConnectionFactoryfactorynewConnectionFactory();factory.setHost(localhost);factory.setPort(5672);factory.setUsername(guest);factory.setPassword(guest);Connectionconnectionfactory.newConnection();Channelchannelconnection.createChannel();// 持久化队列durabletrue第三个参数 exclusivefalse自动删除falsechannel.queueDeclare(QUEUE,true,false,false,null);// 生产一条消息方便直接消费演示channel.basicPublish(,QUEUE,MessageProperties.PERSISTENT_TEXT_PLAIN,order-1001 paid.getBytes(StandardCharsets.UTF_8));// autoAckfalse 是手动确认的关键开关DeliverCallbackdeliverCallback(consumerTag,delivery)-{longtagdelivery.getEnvelope().getDeliveryTag();StringbodynewString(delivery.getBody(),StandardCharsets.UTF_8);try{// 模拟业务真实场景这里是一次数据库写 一次 RPChandleBusiness(body);// 业务成功后才确认multiplefalse 表示只确认这一条channel.basicAck(tag,false);System.out.println([ack] body);}catch(Exceptione){// 业务失败拒绝并重新入队交给其他消费者重试channel.basicNack(tag,false,true);System.err.println([nack-requeue] body causee.getMessage());}};// 注意第二个参数 autoAckfalsechannel.basicConsume(QUEUE,false,deliverCallback,consumerTag-{});System.out.println(consumer started, waiting...);Thread.sleep(10_000);channel.close();connection.close();}privatestaticvoidhandleBusiness(Stringbody){if(body.contains(poison)){thrownewIllegalStateException(业务处理失败);}// 正常业务写库、调用下游等}}运行方式启动本地 RabbitMQ 后javac -cp amqp-client-5.20.0.jar ManualAckDemo.java再java -cp .:amqp-client-5.20.0.jar:slf4j-api-1.7.36.jar ManualAckDemo。预期输出是先打印consumer started随后[ack] order-1001 paid。关键点有三处。第一basicAck放在handleBusiness之后而不是之前这就是「先业务后确认」。第二失败时用了basicNack(tag, false, true)第三个参数requeuetrue表示回队列——但这个选择有风险下一节会专门讲。第三multiplefalse意味着精确确认单条这是初学阶段最安全的默认选择。5. basicQos 预取一次别给我太多上一节的代码有一个隐藏问题Broker 会把队列里的消息尽可能快地推给消费者如果你的消费者只有一个线程处理或者处理速度远低于投递速度unacked 消息会快速堆积。放大到多消费者场景还会出现「忙的消费者手里堆满未确认消息闲的消费者没活干」的倾斜现象。这就是预取prefetch要解决的。basicQos(prefetchCount, prefetchSize, global)是消费者端给 Broker 的限制在同一时间最多允许有多少条「已投递但未确认」的消息挂在我这里。prefetchCount是最常用的参数prefetchSize一般填 0表示不按字节数限制global的语义要特别小心。// 伪代码展示 QoS 的两种作用域不能直接运行// 作用域一仅对之后新建的消费者生效RabbitMQ 常见语义channel.basicQos(10,false/* prefetchSize */,false/* global */);channel.basicConsume(QUEUE,false,deliverCallback,cancelCallback);// 作用域二对当前信道上的所有消费者生效globaltruechannel.basicQos(10,false,true);先记住一个最小模型QoS 不是限速器而是「在途额度」。它限制的是「未确认」的数量不是每秒投递数量。只要你不 ack额度就不会释放。因此prefetchCount1意味着严格的一条一确认、串行处理prefetchCount100意味着允许 100 条同时在途适合高并发批量处理。预取值吞吐特性内存与公平性典型场景1低几乎串行内存占用最小最公平处理慢、单条成本高的任务10~50中等较均衡普通业务消费多数场景的起点100~500高unacked 堆积明显消费者倾斜风险上升批量计算、写入吞吐优先0不限制最高但不可控容易 OOM 并触发流控极少数实验场景不建议生产注意一个反直觉的地方预取值越大单条消息的端到端延迟可能越高。因为消息被推到某个消费者后如果这个消费者在慢慢处理前面的消息后面的消息只能等着而另一个空闲消费者却取不到。所以「提高预取等于提高性能」只在批量、均匀、消费者数量少时成立。6. reject 与 nack拒绝之后的两种命运业务失败时你不能只 ack也不能放着不管。RabbitMQ 给了两个 APIbasicReject和basicNack。它们的关系可以用一句话概括reject 是 nack 的单条版本nack 能批量reject 不能。API能否批量参数常见用途basicAck(tag, multiple)能deliveryTag、multiple业务成功删除消息basicReject(tag, requeue)不能deliveryTag、requeue单条失败明确重投或丢弃basicNack(tag, multiple, requeue)能deliveryTag、multiple、requeue批量失败统一重投或丢弃requeuetrue表示把消息放回队列requeuefalse表示丢弃。丢弃并不等于消息消失得无影无踪——如果队列配置了死信交换机DLX消息会被转发到死信队列供后续人工排查或延迟重试。这里把 DLX 的角色说清楚它是队列的一个属性消息因 nackrequeuefalse、TTL 过期或队列满而被丢弃时Broker 会把它路由到 DLX。一个常见错误是把basicReject(tag, false)当成「重试」。它其实是丢弃。要重试必须requeuetrue但重试次数无法由 RabbitMQ 控制只能靠你自己记录。下面是把「有限重试 死信」串起来的核心片段属于简化代码需配合 DLX 声明才能运行// 简化代码通过 x-death 头判断重试次数超过阈值就丢进死信DeliverCallbackcallback(tag,delivery)-{longdTagdelivery.getEnvelope().getDeliveryTag();MapString,Objectheadersdelivery.getProperties().getHeaders();intretryextractRetryCount(headers);// 从 x-death 头里解析重试次数try{handleBusiness(newString(delivery.getBody(),StandardCharsets.UTF_8));tag.basicAck(dTag,false);}catch(Exceptione){if(retry3){tag.basicNack(dTag,false,true);// 前三次重投}else{tag.basicNack(dTag,false,false);// 超过阈值进死信队列}}};x-death是 Broker 在消息因死信原因被投递时写入的头部记录了死信原因和次数。它的具体字段在不同版本和不同死信原因下略有差异所以解析逻辑要按你实际 Broker 版本验证不要盲抄。如果你不想依赖它更稳妥的做法是在业务侧用一张「消息去重/重试计数表」自己维护。7. 重新入队风险看似安全的 requeuetruerequeuetrue看起来是最省心的失败处理失败了放回去再试嘛。但它藏着两个真实的生产事故源饥饿和风暴。先说饥饿。RabbitMQ 重新入队时默认会把消息放回队列头部附近。如果队列里只有这一条「毒消息」反复失败而你只有一个消费者它会被立刻再次投给同一个消费者导致队列后面所有正常消息被无限期卡住。这就是经典的「毒消息阻塞」。多消费者只能缓解不能根治如果每个消费者都立刻拿到它并失败整条队列的有效吞吐都会下降。再说风暴。当一批消息因为同一个下游故障比如数据库连接池耗尽集体失败requeuetrue会让它们几乎立刻回到队列并再次被投递形成高频重试循环。这不仅把 CPU 和网络打满还会持续刷日志掩盖真正的故障原因。下面这张时序把两种风险串起来毒消息场景(单消费者, requeuetrue): 消费者: 取 msgb - 失败 - nack(requeuetrue) - 重新入队(靠前) 消费者: 又取到 msgb - 失败 - nack(requeuetrue) - 循环... 队列: [b][1][2][3] - 1/2/3 永远轮不到 批量故障场景(下游数据库不可用): msg1..msg100 全部失败 - 全部 requeue - 立刻重新投递 - 再次全部失败 时间轴: 重试间隔约等于 0, 直到数据库恢复或消息被拖入死信正确的工程做法是给重试加上时间和次数约束。时间约束靠延迟队列或 TTL DLX次数约束靠你自己记录的重试计数。当重试超限后用basicNack(tag, false, false)把消息送到死信队列让主队列继续前进。这个组合的完整链路是主队列 - 处理失败且未超限 - 重投 - 超限 - DLX - 死信队列 - 人工或定时补偿。8. 消费者超时与长时间未确认还有一个隐蔽的失败模式消费者既没 ack 也没 nack业务卡在那里。常见原因是下游 RPC 没有超时、数据库慢查询、死锁或线程被阻塞。此时消息一直处于 unacked 状态占用预取额度消费者看起来「活着」但实际不干活。RabbitMQ 有一个连接层面的consumer_timeout消费者确认超时默认 30 分钟可在配置中调整。一旦某条投递在该时间内没有确认Broker 会主动关闭这个信道unacked 消息全部重新入队。这个机制保护的是集群资源不被卡死但它的副作用是如果配置为 30 分钟而你的业务正常处理需要 31 分钟你的消息会被反复重投形成难以排查的重复消费。因此业务处理时间必须明显小于consumer_timeout或者你需要为长任务设计另一种模式先快速 ack 一条「已接收」记录再做异步处理用业务侧台账保证最终一致。很多团队用「先 ack 后处理」来绕过超时但这就把可靠性责任完全转移到了自己身上必须配套幂等和补偿机制否则会丢消息。9. 从一次投递到确认完整链路推演现在让一次真实消费完整走一遍把前面所有概念串起来。场景订单履约消费者预取 20业务是扣库存 发短信下游偶尔超时。[1] Broker 投递 msg(tag7) - 消费者, unacked7 [2] 消费者线程从处理池取线程执行 [3] 扣库存成功 [4] 发短信超时, 抛异常 [5] 判断重试次数 retry0 3 [6] basicNack(7, false, true) - Broker 重新入队 [7] Broker 再次投递 msg(tag9) - 可能给同一或另一消费者 [8] 再次失败... 直到 retry3 [9] basicNack(tag, false, false) - 消息进死信队列 [10] 死信消费者记录工单, 人工介入这个链路里有三个设计决策点。第一在第 6 步和第 9 步之间必须有重试计数否则永远出不去。第二第 7 步的重新投递可能换消费者所以业务必须幂等——同一条消息被处理两次扣库存不能扣两次。第三第 10 步的死信不是终点要有可观测的告警和补偿流程否则死信队列会变成黑洞。下面给出第二个完整示例带幂等键和重试计数的消费者。环境同上额外建议你在 MySQL 里有一张t_msg_log(msg_id, status)表。这里用内存 Map 模拟幂等表保证单进程可运行生产请替换为数据库或 Redis。// IdempotentConsumer.java —— 幂等 有限重试的完整消费者importcom.rabbitmq.client.*;importjava.nio.charset.StandardCharsets;importjava.util.Map;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.atomic.AtomicInteger;publicclassIdempotentConsumer{privatestaticfinalStringQUEUEdemo.order.paid.idem;privatestaticfinalintMAX_RETRY3;// 模拟幂等表生产环境请换成数据库唯一索引或 Redis SETNXprivatestaticfinalMapString,BooleanDONEnewConcurrentHashMap();// 模拟重试计数生产环境请持久化否则重启后计数丢失privatestaticfinalMapString,AtomicIntegerRETRYnewConcurrentHashMap();publicstaticvoidmain(String[]args)throwsException{ConnectionFactoryfnewConnectionFactory();f.setHost(localhost);f.setPort(5672);f.setUsername(guest);f.setPassword(guest);try(Connectionconnf.newConnection();Channelchconn.createChannel()){ch.queueDeclare(QUEUE,true,false,false,null);ch.basicQos(20);// 预取 20控制 unacked 上限DeliverCallbackonDeliver(consumerTag,delivery)-{longdTagdelivery.getEnvelope().getDeliveryTag();StringmsgIddelivery.getProperties().getMessageId();StringbodynewString(delivery.getBody(),StandardCharsets.UTF_8);if(msgIdnull){// 没有幂等键的消息直接拒绝避免无法去重ch.basicNack(dTag,false,false);return;}if(DONE.containsKey(msgId)){// 幂等命中直接确认ch.basicAck(dTag,false);return;}try{handleBusiness(body,msgId);DONE.put(msgId,true);ch.basicAck(dTag,false);}catch(Exceptione){inttimesRETRY.computeIfAbsent(msgId,k-newAtomicInteger()).incrementAndGet();booleanrequeuetimesMAX_RETRY;ch.basicNack(dTag,false,requeue);// 未超限重投超限进死信System.err.println(fail msgIdmsgId timestimes requeuerequeue);}};ch.basicConsume(QUEUE,false,onDeliver,t-{});System.out.println(idempotent consumer started, prefetch20);Thread.sleep(15_000);}}privatestaticvoidhandleBusiness(Stringbody,StringmsgId){// 真实业务扣库存 发短信必须使用 msgId 保证下游可去重System.out.println(handling msgId bodybody);}}预期行为同一条消息即使因 nack 被重投第二次到达时msgId已在DONE里直接 ack不会重复扣库存。失败消息会在重试 3 次后走死信。最容易改错的地方有两个msgId必须在生产端就设置好并全局唯一重试计数如果只放内存进程重启后会归零导致消息无限重试生产必须持久化或改用x-death头。10. QoS、ack 与业务并发怎么配合当消费者内部用线程池并发处理时QoS 的语义会变得微妙。假设预取是 20消费者用 4 个线程处理那么同一时刻最多 20 条在途、最多 4 条在跑其余 16 条在内存队列里排队。这没问题但要注意两点。第一ack 必须在处理线程里调用且要小心 Channel 的线程安全。RabbitMQ Java 客户端的Channel不是线程安全的多线程同时 ack 需要自己加锁或者为每个线程开独立 Channel。很多「偶发 ack 无效」的 bug 都源于此。共享 Channel 上的 deliveryTag 序号是全局递增的批量 ack 时尤其容易误伤。第二预取要略大于并发线程数但不能太大。如果预取等于线程数当某条消息处理特别慢时其他空闲线程就没有新消息可拿吞吐下降如果预取远大于线程数内存里堆积的业务数据会变多且消费者倾斜更严重。一个可用的经验起点是「预取 并发线程数 × 2」再根据压测调整。第三种完整场景是生产端配合消息带messageId队列声明 DLX消费端使用预取与有限重试。下面的命令块展示如何用rabbitmqadmin或管理 API 观察 unacked 与死信情况这属于可复现的排查操作前置条件是开启 management 插件。# 查看队列深度、消费者数、未确认数需要 management 插件与本地 admin 工具rabbitmqadmin list queues name messages messages_unacknowledged consumers# 输出示例真实字段随版本略有差异:# --------------------------------------------------------------------# | name | messages | messages_unacknowledged | consumers |# --------------------------------------------------------------------# | demo.order.paid.idem | 120 | 20 | 1 |# --------------------------------------------------------------------# 查看某个队列的死信配置与消息数rabbitmqadmin list queues name arguments dead_letter_exchangemessages_unacknowledged长期等于预取值通常说明消费者的处理速度跟不上投递速度或者有消息卡住没确认messages持续增长则说明整体消费能力不足。这两个指标是排查消费问题的第一入口。11. 设计取舍可靠、吞吐与重复之间的三角到这里可以把所有取舍收拢成一张决策表。核心矛盾是你要可靠性就要手动确认和重试你要重试就可能重复消费你要高吞吐就要放大预取但放大预取会让失败重投的粒度变粗、内存压力变大。目标推荐配置需要接受的代价宁可重复、不可丢失manual ack 有限重试 死信 幂等业务侧必须实现幂等重复不可避免高吞吐批量处理prefetch 100~500 批量 ack(multipletrue)精度下降失败时可能整批重投严格顺序、低延迟prefetch1 单消费者吞吐低扩展性差长任务先 ack 再异步处理 本地任务表可靠性责任转移到业务侧失败隔离nack(requeuefalse) DLX需要维护死信消费与告警有一个反直觉但重要的结论在分布式系统里「恰好一次」几乎不可能工程上追求的是「至少一次 幂等」。所以与其纠结怎么防止重复不如把幂等做扎实。幂等键可以是业务单号、消息messageId或者由「业务主键 操作类型」拼出来。另一个取舍是批量 ack 的边界。批量 ack 能省网络往返但它把多条消息的命运绑在一起。只有在「同一批消息要么都成功、要么都可安全重试」时才用且必须保证批内不会混入处理结果不同的消息。如果做不到就老老实实单条 ack。12. 生产实践建议与常见误区生产实践建议部分先给几条可以直接落地的规则。第一所有重要业务队列都关闭自动确认使用 manual ack。第二预取值从 20 起步压测后再调不要一上来就设 0。第三失败重试必须有次数上限和时间间隔推荐用 DLX TTL 做延迟重试而不是立刻requeuetrue。第四所有可能重复消费的业务都必须有幂等键和唯一约束。第五为每个消费队列配置死信队列并对死信数量做告警死信为 0 不代表正常突然增长才是信号。常见误区则集中在下面几点误区一以为autoAcktrue只是「自动帮我 ack 一下」。实际上它让消息在投递瞬间就被删除业务失败即丢失。误区二以为basicReject(tag, false)是重试。它是丢弃要重试必须requeuetrue。误区三以为requeuetrue一定会回到队尾。它通常会被放到靠近头部的位置从而阻塞后面的消息。误区四以为 QoS 是限速。它限制的是未确认数量不限制投递速率。误区五在业务成功前就 ack以换取「不重复」。这等于用丢消息换去重非常危险。误区六多个线程共享一个 Channel 并各自 ack。Channel 非线程安全可能引发连接层异常或确认错乱。13. 排障清单从现象到动作现象可能原因排查动作消息莫名消失使用了 autoAck 且业务失败检查消费者autoAck参数改为 falseunacked 长期不降业务卡住、未 ack、超时看messages_unacknowledged检查下游超时与线程栈队列积压但消费者空闲预取过小或消费者倾斜调大预取增加消费者或优化处理速度同一条消息反复重投毒消息 requeuetrue加最大重试次数超限进死信消息被重复消费重投 无幂等引入幂等键与唯一约束连接被 Broker 关闭消费者确认超时检查处理耗时与consumer_timeout配置排查顺序建议是先看队列指标深度、unacked、消费者数再看消费者日志是否 ack、是否异常循环最后看 Broker 日志是否有连接或信道被关闭。不要一上来就改预取值先把「有没有确认」这个根因定位清楚。14. 面试/复盘问题autoAcktrue和autoAckfalse在消息生命周期上的差别是什么分别在什么时机删除消息basicAck的multipletrue有什么语义什么场景下会误伤已经成功的消息basicReject和basicNack的区别是什么什么时候必须用 nackrequeuetrue会带来哪些风险如何设计有限重试basicQos限制的是什么为什么prefetch0通常不建议用于生产消费者超时触发后RabbitMQ 会做什么这对业务处理时间有什么要求为什么说「至少一次 幂等」比追求「恰好一次」更现实你会怎么设计幂等键这些问题可以当作团队评审的检查项如果你的消费者代码答不上第 2、4、7 题说明它在失败路径上的行为是不确定的。15. 总结把全文收回到一张框架图生产端负责让消息可路由、可去重Broker 负责排队、投递和在未确认时保留消息消费端负责在业务成功后确认、失败时受控重试、超限进死信。手动确认解决「丢消息」预取解决「堆积与倾斜」nack/reject 解决「失败怎么处置」死信与幂等解决「重试之后怎么办」。最后给一条可以直接执行的判断规则先关闭 autoAck、把预取设为 20、给失败加最大重试次数、给业务加幂等键、给队列加死信。这五步做完你就已经避开了绝大多数消费端事故。剩下的调优——预取调多大、并发开多少、批量 ack 用不用——都可以在压测中慢慢找答案。16. 参考资料RabbitMQ 官方文档Consumer Acknowledgements and Publisher ConfirmsRabbitMQ 官方文档Confirms (Publisher Confirms) 与 Consumer PrefetchRabbitMQ 官方文档Dead Lettering、TTL 与 Queue Length LimitRabbitMQ 官方文档Consumer Timeoutconsumer_timeout配置说明RabbitMQ Java Client API 文档Channel、DeliverCallback、basicAck、basicNack、basicQosAMQP 0-9-1 规范amqp.org《RabbitMQ 实战指南》相关章节朱忠华