RabbitMQ/Spring AMQP 如何实现队列消息留存 支持消费异常后重新消费

你需要的"业务操作（如DB写入）出现异常时，消息仍可被重新消费"的能力，完全可以基于RabbitMQ原生的消息确认机制实现，不需要自行实现消息留存的自定义逻辑，核心思路是避免框架在业务逻辑执行完成前，就将消息从队列中移除。

核心配置逻辑 #

• 关闭框架自动确认模式，将RabbitMQ监听器的应答模式调整为手动确认，对应配置项为spring.rabbitmq.listener.simple.acknowledge-mode: MANUAL
• 切换为手动模式的原因：默认的AUTO模式是监听器方法执行完就自动给RabbitMQ发确认指令，哪怕方法内触发的异步DB写入后续执行失败，消息已经被从队列删除，根本没有重回的机会；而MANUAL模式把确认消息的时机完全交给业务代码控制，刚好匹配"业务成功才移除消息、失败就退回消息"的需求
• 可搭配配置spring.rabbitmq.listener.simple.prefetch: 1，限制单次给消费者推送的消息数量，避免消费积压时短时间拉取大量消息占用本地资源
• 该模式下，RabbitMQ推送消息给消费者后，不会自动把消息从队列标记为已删除，只有收到消费者的明确确认指令，才会真正移除队列里的消息

监听器内的业务逻辑写法 #

按照实际业务执行结果做明确的消息应答：

• 业务执行成功（比如DB数据写入完成无异常）：手动调用确认接口channel.basicAck(deliveryTag, false)，告知RabbitMQ这条消息已经处理完成，可以从队列移除
• 业务执行抛出异常：调用消息否定接口channel.basicNack(deliveryTag, false, true)，最后一个参数requeue设为true，RabbitMQ会将这条消息重新放回原队列的尾部，等待后续再次被消费

示例代码：

@RabbitListener(queues = "your.biz.queue")
public void handleBizMessage(Message message, Channel channel) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    try {
        // 执行业务逻辑，例如DB写入
        saveOrderToDatabase(message);
        // 业务成功，确认消息移除
        channel.basicAck(deliveryTag, false);
    } catch (Exception ex) {
        // 业务异常，消息重回原队列等待重新消费
        channel.basicNack(deliveryTag, false, true);
    }
}

常见踩坑提醒 #

• 不要直接依赖"消费者异常断开连接后消息自动重回队列"的机制：这种方式会导致RabbitMQ长时间维护未确认消息的状态，占用服务端内存资源，远不如主动在异常捕获流程中调用nack接口的效率高
• 一定要做业务幂等设计：重回队列的消息可能被多次消费，DB写入逻辑需要增加唯一约束、去重判断，避免重复写入产生脏数据
• 必须配置兜底机制：如果是消息本身存在问题（例如格式错误、永久无法处理），无限重回队列会导致消息在队列中循环堵塞，挤占正常消息的消费资源。可以搭配Spring Retry配置重试次数阈值，超过阈值的消息直接转入死信队列，不再重回原队列
• 注意区分nack的参数：第二个参数multiple要设为false，避免把之前未确认的所有消息一并处理；第三个参数requeue要按需设置，不希望重回的场景要把该值设为false，让消息走死信流程

内容的提问来源于stack exchange，提问作者user5822650