当前位置：首页 > news >正文

RabbitMQ 如何保证消息不丢失和不重复消费？掌握这 4 个关键点

news 2026/6/14 0:27:38

在开发中，我们经常用RabbitMQ来做系统之间的传话筒。

比如用户下单后，通知库存系统减库存、通知物流系统准备发货。

但问题来了：万一消息丢了怎么办？或者同一条消息被处理了两次怎么办？

别担心！只要做好以下几点，就能让 RabbitMQ 变得既可靠又安全。

消息可能在哪丢？

假设发快递：

你（生产者）把包裹交给快递员（RabbitMQ）；
快递员把包裹送到收件人（消费者）手上。

在这个过程中，包裹可能在三个地方出问题：

你刚寄出，快递员没收到→ 消息没到 RabbitMQ；
快递员收到了，但仓库门没开→ RabbitMQ 宕机，消息没了；
收件人签收前手机没电了→ 消费者处理失败，消息丢失。

所以，我们要从发送方、中间方、接收方三处下手！

先配好配置文件（application.yml）

yaml

spring: rabbitmq: host: localhost port: 5672 username: guest password: guest # 开启 publisher confirm 和 return publisher-confirm-type: correlated publisher-returns: true template: mandatory: true # 使 returns 生效 listener: simple: acknowledge-mode: manual # 手动 ACK retry: enabled: false # 我们自己控制重试逻辑 redis: host: localhost port: 6379

1. 声明队列、交换器（持久化！）

创建队列或交换器时，设置durable=true队列持久化。

java @Configuration public class RabbitMqConfig { public static final String ORDER_EXCHANGE = "order.exchange"; public static final String ORDER_QUEUE = "order.queue"; public static final String ORDER_ROUTING_KEY = "order.create"; @Bean public DirectExchange orderExchange() { // durable = true（默认就是 true） return new DirectExchange(ORDER_EXCHANGE, true, false); } @Bean public Queue orderQueue() { // durable = true return new Queue(ORDER_QUEUE, true); } @Bean public Binding binding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()) .with(ORDER_ROUTING_KEY); } }

2. 生产者发送消息（带唯一 ID + Confirm 回调）

发送消息时，设置 deliveryMode=2 消息持久化。

同时增加异步非阻塞操作，发完消息立刻返回，RabbitMQ 后台异步确认。支持批量确认、单条确认。

java @Service public class OrderProducer { @Autowired private RabbitTemplate rabbitTemplate; // 生成唯一消息ID（实际可用 UUID 或业务ID） public void sendOrderMessage(String orderId) { String msgId = "msg_" + System.currentTimeMillis(); // 简化版唯一ID MessageProperties props = new MessageProperties(); props.setMessageId(msgId); // 设置唯一ID，用于幂等 props.setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 持久化 Message message = new Message((orderId).getBytes(StandardCharsets.UTF_8), props); // 发送并监听 confirm rabbitTemplate.convertAndSend( RabbitMqConfig.ORDER_EXCHANGE, RabbitMqConfig.ORDER_ROUTING_KEY, message, new CorrelationData(msgId) // CorrelationData 用于关联 confirm ); // 设置 confirm 回调 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (ack) { System.out.println("消息已确认送达 RabbitMQ: " + correlationData.getId()); } else { System.err.println("消息发送失败: " + cause); // 可在此处记录日志、重发、存 DB 等 } }); // （可选）设置 return 回调，处理路由失败 rabbitTemplate.setReturnsCallback(returned -> { System.err.println("消息无法路由: " + new String(returned.getMessage().getBody())); }); } }

生产环境建议把失败的消息存入数据库，由定时任务补偿重发。

到这里很多人以为只要设置了durable=true和deliveryMode=2，消息就万无一失了。

其实不然！RabbitMQ 收到持久化消息后，会先写入内存缓冲区，再异步刷盘（fsync）。

如果在这之间服务器断电，消息还是会丢！

解决方案：

传统方案：镜像队列（Mirrored Queue）多节点备份，但存在脑裂、数据不一致风险。

现代方案（RabbitMQ 3.8+）：Quorum Queue（仲裁队列）

基于 Raft 共识算法，强一致性；
自动选主、故障转移；
写入多数节点才返回成功，真正防丢。

在SpringBoot中声明 Quorum Queue：

java

@Bean public Queue quorumOrderQueue() { return QueueBuilder .durable("order.quorum.queue") .quorum() // 关键！ .build(); }

3.消费者：手动 ACK + 幂等处理（防重复）

手动 ACK 是什么？等待消费者调用basicAck，收到 ACK 后，才从队列中删除消息。

什么是幂等？一个操作执行一次和执行多次，结果完全相同。

java @Component public class OrderConsumer { @Autowired private StringRedisTemplate redisTemplate; private static final String CONSUMED_KEY_PREFIX = "mq:consumed:"; @RabbitListener(queues = RabbitMqConfig.ORDER_QUEUE) public void handleOrder(Message message, Channel channel) throws IOException { String msgId = message.getMessageProperties().getMessageId(); String orderId = new String(message.getBody(), StandardCharsets.UTF_8); try { // 1.幂等检查：是否已处理过？ String key = CONSUMED_KEY_PREFIX + msgId; Boolean hasConsumed = redisTemplate.opsForValue().setIfAbsent(key, "1", Duration.ofHours(24)); if (Boolean.FALSE.equals(hasConsumed)) { System.out.println("重复消息，跳过处理: " + msgId); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); return; } // 2.处理业务逻辑（比如减库存、发短信） System.out.println("正在处理订单: " + orderId); // 模拟业务耗时 Thread.sleep(1000); // 3.业务成功 → 手动 ACK channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); System.out.println("消费成功，已 ACK: " + msgId); } catch (Exception e) { System.err.println("消费失败: " + e.getMessage()); // 拒绝消息，不 requeue（避免死循环），或根据策略决定是否重试 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); // 或者：basicReject + 记录到死信队列 } } }

关键点：

setIfAbsent 实现 Redis 分布式锁式去重；
先检查幂等，再处理业务，最后 ACK；
异常时用 basicNack 拒绝消息，避免无限重试。

完整可靠性链路图

ini

[生产者] │ ├── 开启 Confirm → 确保消息到达 Broker └── 消息带唯一ID → 用于后续幂等 │ ▼ [RabbitMQ Broker] ├── 队列/交换器持久化 ├── 消息持久化（deliveryMode=2） └── 使用 Quorum Queue（高可用+强一致） │ ▼ [消费者] ├── 手动 ACK（autoAck=false） ├── 先查幂等（Redis/setIfAbsent） ├── 再执行业务 └── 最后 ACK（失败则 NACK 或进死信队列）

常见误区

误区	正确做法
“开了持久化就不会丢”	还需 Confirm + 高可用队列
“自动 ACK 更简单”	自动 ACK 极易丢消息！必须手动
“RabbitMQ 能保证不重复”	不能！必须消费者自己幂等
“消息ID用时间戳就行”	时间戳可能重复！建议用 UUID 或雪花ID