# RabbitMQ上线3天丢了1200条消息，我把生产者、MQ、消费者三处都重写了

> 👋 我是小坏，一个天天写Java的普通开发者。每天分享写代码、跑线上踩出来的真经验，不废话。

上周三刚上线的订单异步通知模块，周一早上产品跑过来问："怎么1200多笔订单没通知下游系统？"

我查了下监控，消息发送成功率99.6%，看起来挺正常。一翻日志吓一跳：那0.4%正好是周末业务高峰丢的——更要命的是，下游没收到消息就没发货，用户已经开始投诉了。

这事儿从头到尾踩了RabbitMQ最经典的四个坑，我把三处代码全重写了一遍。说点真东西，网上教程很少有人一次把这四个坑讲全。

## 坑一：生产者发消息，没开confirm，发出去就当成功了

我一开始写的代码特别自信：

```java
rabbitTemplate.convertAndSend("order-exchange", "order.create", order);
```

这行代码返回值是void。也就是说，你调完它那一刻，只代表消息写到了Socket缓冲区，**不代表Broker真的收到了**。

我线上遇到的情况是：MQ节点短暂重启了200ms，Socket没报错但消息没进Broker，生产者这边完全没感知。1200条消息里大概200条是这么丢的。

修法是开publisher confirm + return callback，并且生产者同步等待broker ack，超过3秒没ack就重试3次：

```yaml
spring:
  rabbitmq:
    publisher-confirm-type: correlated  # 开启发送确认
    publisher-returns: true              # 开启消息回退
    template:
      mandatory: true                    # 路由失败时退回，不要静默丢
```

发送代码改成这样：

```java
public void sendWithConfirm(Order order) {
    CorrelationData correlationData = new CorrelationData(order.getOrderId());
    rabbitTemplate.convertAndSend("order-exchange", "order.create", order, correlationData);
    try {
        // 等待broker ack，最多3秒
        CorrelationData.Confirm confirm = correlationData.getFuture().get(3, TimeUnit.SECONDS);
        if (!confirm.isAck()) {
            log.warn("消息未被broker ack，重试中: orderId={}", order.getOrderId());
            retry(order);
        }
    } catch (Exception e) {
        log.error("等待ack超时，重试: orderId={}", order.getOrderId(), e);
        retry(order);
    }
}
```

这里我个人觉得有个坑，`getFuture().get()`默认是无限等待的，一定要加超时时间，不然MQ宕机你这里线程全卡死。

## 坑二：消息到了Exchange但路由不到Queue，也会丢

一开始我以为开了publisher confirm就万事大吉。后来看RabbitMQ官方文档才知道：**confirm只保证消息到达了Exchange，不保证进了Queue**。

如果routing key写错了、Queue没绑定、Queue被人删了，消息会被直接丢弃，这时候confirm仍然返回true。

我线上就踩了这个——有个哥们改了Queue名字没改消费者那边的binding key，大概100条消息路由到不存在的Queue，悄无声息地没了。

修法是开publisher returns，并记录日志+告警：

```java
@Slf4j
@Configuration
public class RabbitConfig implements RabbitTemplate.ReturnsCallback {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
        RabbitTemplate t = new RabbitTemplate(factory);
        t.setReturnsCallback(this);
        return t;
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("消息被退回，未路由到任何Queue! exchange={}, routingKey={}, replyText={}",
                returned.getExchange(), returned.getRoutingKey(), returned.getReplyText());
        // 告警：发钉钉/企微
        alert("消息路由失败: " + returned.getRoutingKey());
    }
}
```

这一步做完，消息"凭空消失"的情况基本就没了——只要消息没进Queue，你第一时间能知道。

## 坑三：队列和消息都没持久化，Broker一重启全没

我最初创建队列用的是：

```java
// 错误示例！durable默认false
@Bean
public Queue orderQueue() {
    return new Queue("order-queue");  // 这是非持久的
}
```

消息也没设deliveryMode，默认是非持久。结果周末运维做机器迁移重启了一次Broker，内存里大概700条未消费的消息，重启完全没了。这是丢得最多的一次。

修法是三处都要改：

```java
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order-queue")      // 队列持久化
            .withArgument("x-dead-letter-exchange", "dlx-exchange")
            .withArgument("x-dead-letter-routing-key", "dead")
            .build();
}

@Bean
public DirectExchange orderExchange() {
    return ExchangeBuilder.directExchange("order-exchange")
            .durable(true)                         // 交换机持久化
            .build();
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
    RabbitTemplate t = new RabbitTemplate(factory);
    t.setReturnsCallback(this);
    // 消息持久化：deliveryMode=2
    t.setMessageConverter(jackson2JsonMessageConverter());
    return t;
}
```

发消息时还要显式设delivery mode（很多转换器默认非持久，这点要注意）：

```java
MessageProperties props = MessagePropertiesBuilder.newInstance()
        .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
        .setMessageId(order.getOrderId())
        .build();
Message message = new Message(json.getBytes(), props);
rabbitTemplate.send("order-exchange", "order.create", message, correlationData);
```

三点缺一个都不行：交换机持久、队列持久、消息持久。我当时只设了队列持久，消息还是在重启时丢了，浪费了两个小时排查。

## 坑四：消费者auto ack，消息还没处理完就被删了

消费者那边我一开始写的是：

```java
@RabbitListener(queues = "order-queue")
public void onMessage(Order order) {
    notifyDownstream(order);  // 如果这里抛异常呢？
}
```

默认模式是auto ack——消息投递给消费者那一刻就被MQ删掉了。如果notifyDownstream()抛异常，或者消费者进程在处理中崩了，这条消息就彻底没了。

我线上大概200条是这么丢的：下游接口超时抛异常，消息已经被删，再也不会重试。

改成手动ack：

```yaml
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual   # 手动ack
        prefetch: 10                # 每次最多预取10条
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000ms
```

消费者：

```java
@RabbitListener(queues = "order-queue")
public void onMessage(String body, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        Order order = JSON.parseObject(body, Order.class);
        notifyDownstream(order);
        channel.basicAck(tag, false);              // 处理成功才ack
    } catch (Exception e) {
        log.error("处理消息失败，重回队列", e);
        channel.basicNack(tag, false, true);       // 失败nack，requeue=true重试
    }
}
```

这里有个细节：nack 加 requeue=true 如果一直失败会无限重试打日志，所以我配合了死信队列——重试超过5次自动转到死信队列告警，不会一直堵着。

## 最后再加一层兜底：定时对账

做完上面这些，我还是不敢完全信任。毕竟代码写得再严谨，也会有边缘case：数据库字段长度不够、序列化失败、第三方类库有bug……

我加了个每小时跑一次的定时任务，对比订单表和消息发送记录，发现有"订单已支付但消息未发送成功"的记录就补一条：

```java
@Scheduled(cron = "0 0 * * * ?")
public void reconcile() {
    // 查1小时内已支付但无消息记录的订单
    List<Order> missed = orderMapper.selectPaidWithoutMsg(LocalDateTime.now().minusHours(2), LocalDateTime.now().minusMinutes(5));
    for (Order order : missed) {
        log.warn("对账发现消息缺失，补发: orderId={}", order.getOrderId());
        try {
            sendWithConfirm(order);
            msgRecordMapper.markSent(order.getOrderId());
        } catch (Exception e) {
            log.error("补发失败: orderId={}", order.getOrderId(), e);
            alert("消息补发失败: " + order.getOrderId());
        }
    }
}
```

## 改动前后对比

| 维度 | 改动前 | 改动后 |
|---|---|---|
| 3天丢消息数 | 约1200条 | 0条 |
| 生产者confirm | 无 | correlated+3秒超时+3次重试 |
| 消息路由失败 | 静默丢弃 | returnsCallback + 告警 |
| 队列/消息持久化 | 都没设 | 交换机/队列/消息三处全持久 |
| 消费者ack | auto | 手动ack + 死信队列 |
| 兜底 | 无 | 每小时定时对账补发 |

**一句话总结：** 别迷信任何MQ"默认安全"，生产者要等ack、路由失败要回退、队列消息要持久、消费者要手动ack，最后再加个对账兜底。

我个人觉得这四层里，定时对账是最容易被忽略但最救命的——其他三层都是"防"，对账是"查"，防不住的问题总能查到。这个方案不一定适合所有场景，超高频业务（比如秒级十万条消息）对账压力会很大，可以改成抽样对账或者离线跑，核心思路是一样的。

---

## 👋 关于我

我是小坏，天天写Java代码的普通开发者，写代码5年多了。写教程写着写着，发现市面上的工具用着实在闹心——刷题APP广告满天飞，简历模板工具瞎凑经历、改个排版比写代码还累。干脆自己撸了俩工具，把攒了5年的东西全塞进去。

### 📱 小坏面试（微信小程序）

**微信小程序**

5年大厂Java面试题+AI模拟面试，无广告

🔍 微信搜「小坏面试」

### 🌐 小坏简历（在线工具）

**在线工具**

上传旧简历一键诊断，贴JD自动生成适配简历

✨ 一键诊断简历 → https://ai.e404e.cn

---

**📌 关注「小坏说Java」**

每天一篇技术干货+程序员职场观察。刷题、改简历、学技术，一站搞定。

👋 我是小坏，一个天天写Java的普通开发者。每天分享写代码、跑线上踩出来的真经验，不废话。

RabbitMQ上线3天丢了1200条消息，我把生产者、MQ、消费者三处都重写了

编程语言

RabbitMQ上线3天丢1200条消息？从生产者到消费者，4个致命配置陷阱全解析。开启confirm+return回调、三处持久化、手动ack加死信队列，最后用定时对账兜底，彻底告别消息丢失。

代码解释

数据库

路由

JSON

Java

让每一个 Agent，都跑在云上

trabbit

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

RabbitMQ上线3天丢了1200条消息，我把生产者、MQ、消费者三处都重写了-腾讯云开发者社区-腾讯云

RabbitMQ上线3天丢了1200条消息，我把生产者、MQ、消费者三处都重写了

RabbitMQ上线3天丢了1200条消息，我把生产者、MQ、消费者三处都重写了

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐