RabbitMQ 消息可靠性怎么避免消息丢失目录消息可能丢在哪生产者确认持久化消费者手动ACK死信队列幂等消费完整可靠性方案小结笔者主页文章《RabbitMQ 交换机四种交换机理解消息的路由规则》介绍了 RabbitMQ 的交换机和路由机制解决的是消息如何找到正确的队列。但消息成功发送出去并不代表一定能被可靠消费。生产环境中网络异常、服务宕机等情况都可能导致消息丢失。发了一条订单已支付的消息给库存系统库存系统没收到订单显示已支付库存没扣减。这种情况会导致订单状态和库存状态不一致是生产环境中需要重点避免的问题。消息从生产者到消费者中间经过的每一个环节都可能丢消息。要保证消息不丢需要覆盖整个链路。消息可能丢在哪消息的生命周期分三段每段都有丢失风险环节丢失原因后果生产者 → Broker网络抖动、Broker宕机消息根本没进队列Broker 存储队列和消息没做持久化Broker重启消息进过队列但重启后没了Broker → 消费者自动ACK模式下消费者拿到消息就崩溃消息被标记为已消费实际没处理生产者确认生产者把消息发给 Broker怎么知道 Broker 确实收到了默认情况下生产者发完就不管了Broker 有没有收到它不知道。网络抖一下消息就丢了生产者还觉得自己发成功了。RabbitMQ 提供了Publisher Confirm机制生产者发消息后Broker 会给一个确认回调告诉生产者我收到了或者出了问题。Spring Boot 里开启确认机制需要配置publisher-confirm-typespring:rabbitmq:publisher-confirm-type:correlatedpublisher-confirm-type有三个值值行为适用场景none不确认默认不关心消息是否到达Brokercorrelated异步回调每条消息单独确认生产环境推荐simple同步等待确认性能差仅调试用correlated模式下每条消息附带一个唯一 IDBroker 确认后通过回调函数通知生产者。如果 Broker 返回 nack拒绝说明消息没成功投递。生产环境中通常不会直接无限重试而是记录失败消息通过补偿任务或人工介入处理。ServicepublicclassReliableProducer{AutowiredprivateRabbitTemplaterabbitTemplate;PostConstructpublicvoidinitConfirmCallback(){// 确认回调rabbitTemplate.setConfirmCallback((correlationData,ack,cause)-{if(ack){System.out.println(消息到达Broker: correlationData.getId());}else{// 消息没到达Broker需要重试或记录System.err.println(消息投递失败: correlationData.getId(), 原因: cause);}});}publicvoidsendOrderMessage(StringorderId){// 消息唯一IDCorrelationDatacorrelationDatanewCorrelationData(orderId);rabbitTemplate.convertAndSend(order.exchange,order.pay,订单已支付: orderId,correlationData);}}注意Publisher Confirm 只保证消息到达了 Broker 的交换机不保证消息进了队列。如果 routing key 写错了消息到了交换机但没匹配到任何队列confirm 回调依然会返回 ack因为交换机收到了但消息被丢弃了。要确认消息是否进了队列还需要额外设置mandatory参数和ReturnsCallbackPostConstructpublicvoidinitReturnCallback(){// 消息无法路由到队列时触发回调rabbitTemplate.setReturnsCallback(returned-{System.err.println(消息被退回: returned.getMessage(), 交换机: returned.getExchange(), routingKey: returned.getRoutingKey(), 原因: returned.getReplyText());});}publicvoidsendOrderMessage(StringorderId){rabbitTemplate.setMandatory(true);// 消息无法路由时退回rabbitTemplate.convertAndSend(order.exchange,order.pay,订单已支付: orderId);}mandatory true告诉 Broker如果这条消息找不到匹配的队列不要默默丢掉退回来告诉我。持久化消息到了 Broker存在内存里。Broker 一重启内存清空消息全没了。要让消息扛住重启需要做持久化。持久化涉及三个对象三个都要设只设一个不够1. 交换机持久化声明交换机时设置durable trueBeanpublicDirectExchangeorderExchange(){// 持久化交换机returnnewDirectExchange(order.exchange,true,false);}2. 队列持久化声明队列时设置durable trueBeanpublicQueueorderQueue(){// 持久化队列returnnewQueue(order.queue,true);}3. 消息持久化发送消息时设置投递模式为持久化rabbitTemplate.convertAndSend(order.exchange,order.pay,message,msg-{// 持久化消息msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);returnmsg;});Spring AMQP 默认发送的消息就是持久化的但显式设置更稳妥。为什么三个都要设想一个场景队列设了持久化消息没设持久化。Broker 重启后队列恢复了但队列里的消息没了。反过来消息设了持久化队列没设持久化。Broker 重启后队列本身都没了消息往哪恢复持久化的代价是性能。每条消息都要写磁盘吞吐量会下降。如果业务场景允许少量丢失比如日志采集可以不做强持久化。但对于支付、订单这类关键业务持久化是底线。消费者手动ACK消费者从队列拿到消息后需要告诉 Broker “我处理完了”这个动作叫 ACKAcknowledgement。RabbitMQ 收到 ACK 后才会把消息从队列里删掉。问题在于Spring Boot 默认是自动 ACK。消费者一拿到消息RabbitMQ 就认为消费成功了立刻删除。如果消费者拿到消息后还没处理完就崩了消息已经从队列删了丢了。spring:rabbitmq:listener:simple:acknowledge-mode:manual# 改为手动确认手动 ACK 模式下消费者处理完消息后显式调用basicAck处理失败调用basicNackComponentpublicclassOrderConsumer{RabbitListener(queuesorder.queue)publicvoidhandleOrder(Messagemessage,Channelchannel)throwsIOException{// 拿到消息的投递标签用于ACKlongdeliveryTagmessage.getMessageProperties().getDeliveryTag();try{// 处理业务逻辑StringbodynewString(message.getBody());processOrder(body);// 处理成功确认消息// multiple false只确认这一条channel.basicAck(deliveryTag,false);}catch(Exceptione){// 处理失败拒绝消息// requeue false不重新入队交给死信队列处理channel.basicNack(deliveryTag,false,false);}}privatevoidprocessOrder(Stringbody){// 实际业务逻辑System.out.println(处理订单: body);}}basicNack的第三个参数requeue很关键requeue行为适用场景true消息重新放回队列头部下次还会被消费临时性错误比如数据库连接超时false消息不回队列如果有死信队列就进死信队列没有就被丢弃确定性错误比如消息格式不对设requeue true要小心如果消息本身有缺陷比如格式错误每次消费都失败每次都被放回队列然后又被消费又失败陷入死循环。这种情况应该设requeue false把消息扔进死信队列由人工或专门的消费者处理。死信队列消费失败的消息、过期没人消费的消息、队列满了溢出的消息这些消息不能直接丢掉得有个地方收容。这个地方叫死信队列Dead Letter QueueDLQ。消息变成死信有三种情况来源触发条件示例消费者拒绝basicNack 或 basicRejectrequeue false消息格式错误处理不了TTL 过期消息或队列设置了 TTL超时未被消费订单30分钟未支付超时消息队列满了队列达到最大长度新消息把旧消息挤出去队列设了 x-max-length声明死信队列和普通队列一样关键是在普通队列上绑定死信交换机ConfigurationpublicclassDlxConfig{// 死信交换机BeanpublicDirectExchangedlxExchange(){returnnewDirectExchange(dlx.exchange);}// 死信队列BeanpublicQueuedlxQueue(){returnnewQueue(dlx.queue,true);}// 死信绑定BeanpublicBindingdlxBinding(){returnBindingBuilder.bind(dlxQueue()).to(dlxExchange()).with(dlx.routing);}// 业务队列指定死信交换机BeanpublicQueueorderQueue(){MapString,ObjectargsnewHashMap();// 绑定死信交换机args.put(x-dead-letter-exchange,dlx.exchange);// 死信消息在死信交换机上的routing keyargs.put(x-dead-letter-routing-key,dlx.routing);returnnewQueue(order.queue,true,false,false,args);}}当order.queue里的消息被 reject 且 requeue false 时RabbitMQ 会自动把这条消息转发到dlx.exchange再路由到dlx.queue。你可以在死信队列上写一个消费者专门处理这些出了问题的消息记录日志、告警通知、人工排查。死信队列用于保存无法正常处理的消息方便后续排查和补偿。正常流程的消息走正常队列出了问题的消息走死信队列两套通道互不干扰。幂等消费前面的机制保证了消息大概率不丢但还有一个问题消息可能被重复投递。场景消费者拿到消息处理成功但在调basicAck之前网络断了。RabbitMQ 没收到 ACK认为消费失败把消息重新投递给另一个消费者。同一条消息被处理了两次。如果消费逻辑是扣库存扣两次就出事了。所以消费端需要做幂等处理同一个操作执行一次和执行多次结果相同。一种常见做法是使用消息唯一 ID 进行去重ComponentpublicclassIdempotentConsumer{AutowiredprivateRedisTemplateString,StringredisTemplate;RabbitListener(queuesorder.queue)publicvoidhandleOrder(Messagemessage,Channelchannel)throwsIOException{// 消息唯一IDStringmessageIdmessage.getMessageProperties().getMessageId();Stringkeyconsumed:messageId;// SETNX判断是否已消费过BooleanfirstTimeredisTemplate.opsForValue().setIfAbsent(key,1,24,TimeUnit.HOURS);if(Boolean.FALSE.equals(firstTime)){// 已经消费过了直接确认不做业务处理channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);return;}try{processOrder(newString(message.getBody()));channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch(Exceptione){// 消费失败删除去重标记允许重试redisTemplate.delete(key);channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);}}}幂等消费的具体方案在之前的《接口幂等性》那篇里详细讲过这里不再展开。实现方式都是用唯一标识判断是否重复重复就跳过。完整可靠性方案把上面五个机制串起来一张图看完整的消息可靠性链路生产者 │ ├── 1. Publisher Confirm ──→ 确认消息到达Broker │ ▼ RabbitMQ Broker │ ├── 2. 交换机持久化 队列持久化 消息持久化 ──→ 重启不丢数据 │ ▼ 消费者 │ ├── 3. 手动ACK ──→ 处理完才确认崩溃了消息还在队列里 │ ├── 处理失败 ──→ basicNack(requeuefalse) ──→ 死信队列兜底 │ └── 4. 幂等消费 ──→ 重复投递不会重复处理机制解决哪个环节的丢失实现方式Publisher Confirm生产者 → BrokerBroker 收到后回调确认持久化Broker 存储交换机、队列、消息三者持久化手动 ACKBroker → 消费者处理完才确认否则消息保留死信队列消费失败的消息失败消息进死信队列等待处理幂等消费重复投递同一条消息只处理一次小结RabbitMQ 的可靠性需要从生产、存储、消费三个阶段分别处理。Confirm 解决生产端确认问题持久化保证 Broker 重启后的数据恢复手动 ACK 和幂等消费保证消费过程稳定死信队列为处理失败的消息提供兜底。但这套方案也有代价Publisher Confirm 多了一次网络往返持久化多了磁盘 IO手动 ACK 增加了消费端复杂度死信队列需要额外的监控和处理逻辑。可靠性越高吞吐量越低系统越复杂。如果你的场景是日志采集、行为埋点这种允许少量丢失的业务没必要上全套。但如果是支付、订单、库存这类关键业务这几个机制是最低配置。
