企业官网建设流程全解析

帮客户落地RabbitMQ：3个可靠性投递与重复消费实战要点

前不久帮一个做招聘SaaS的客户搞了个项目，他们的核心业务是职位信息同步。客户那边有几个数据源，每天要定时从不同的招聘网站把最新的职位数据同步到自己的平台上，同时还要把平台上的职位状态变更推送给外部系统。原来的方案是用数据库轮询+Kafka，但运维成本和网络开销扛不住了，特别是网络抖动频繁导致数据不一致。客户希望引入RabbitMQ来解耦，要求消息不能丢，而且不能重复消费。项目规模大概是每天处理50万条消息，峰值3000条/秒左右。

可靠性投递：生产端确认机制

一开始我觉得RabbitMQ设个手动ACK就完事了，结果发现生产端才是第一道鬼门关。测试的时候模拟了断网、Broker宕机，发现消息在发送过程中就丢了，消费者那边连个影子都没见到。当时有两种选择：事务机制和publisher confirm机制。RabbitMQ的事务确实能保证不丢消息，但性能损耗太大了，实测TPS直接掉到原来的三分之一还多。我最后选了publisher confirm+异步回调的方案，虽然代码复杂度上去了，但性能和可靠性都能兼顾。

核心代码大概长这样：

```java
// 开启确认模式
channel.confirmSelect();

// 发送消息的回调
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
// 确认成功：从本地缓冲中移除
if (multiple) {
// 批量确认
messageBuffer.headMap(deliveryTag + 1).clear();
} else {
messageBuffer.remove(deliveryTag);
}
}

@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
// 确认失败：从本地缓冲取出重发
Map failed = multiple ?
messageBuffer.headMap(deliveryTag + 1) :
Map.of(deliveryTag, messageBuffer.get(deliveryTag));
retryFailedMessages(failed);
}
});
```

这里有个差点踩进去的坑：批量确认模式下handleAck里的deliveryTag是最后一个确认的消息编号，我当时直接按单个消息去清理缓冲区，结果一堆消息被重复确认了。查了两天才意识到需要用headMap截断。

重复消费：幂等性设计是最后防线

消费者端重复消费的问题更恶心。客户那边的业务场景是这样的：用户提交简历后，系统要异步发送邮件通知、更新数据库状态、记录操作日志。如果一条消息被消费两次，数据库状态会脏，邮件会重复发。

有意思的是，网上很多人说用Redis分布式锁就能解决，但我试了一圈发现，在消息量大的时候Redis锁的过期时间很难控制。锁太短会失效，锁太长性能又不行。最后我选了业务层幂等设计，用数据库唯一索引+本地消息表来兜底。

具体做法是：消息体里带一个全局唯一的消息ID（用UUID就行），消费者先查本地消息表，如果这个ID已经处理过就直接ACK跳过。同时数据库里对“业务唯一标识+业务类型”建联合唯一索引，双重保险。

```java
// 消费者幂等判断
public boolean isDuplicate(String messageId, String businessType) {
// 使用数据库唯一索引做幂等
// 表结构：message_id business_type status
return duplicateCheckService.exist(messageId, businessType);
}

// 处理消息
public void handleMessage(MessageData data) {
if (isDuplicate(data.getMessageId(), data.getBusinessType())) {
// 重复消息，直接确认
channel.basicAck(envelope.getDeliveryTag(), false);
return;
}

// 执行业务逻辑
try {
processBusinessLogic(data);
// 记录消息处理日志
duplicateCheckService.save(data.getMessageId(), data.getBusinessType());
channel.basicAck(envelope.getDeliveryTag(), false);
} catch (Exception e) {
// 业务失败，进入重试逻辑
channel.basicNack(envelope.getDeliveryTag(), false, true);
}
}
```

坑死了的一件事是：一开始我把消息ID放到消息头部，消费者出来后再组装。结果有一次消费者代码把消息头部的类型搞混了，所有消息都被判断成重复。老老实实把ID放到消息体里，用JSON解析，虽然多了一点序列化开销，但至少不会出这种低级错误。

可靠重试：死信队列配合TTL

还有一个问题是消息处理失败后的重试策略。客户要求：普通失败重试3次，每次间隔5分钟；连续失败后进入死信队列，人工介入。

我一开始用RabbitMQ的重试插件，但在3.13版本下发现插件和Spring Boot的集成有点发神经，某些情况下回导致消费者无限重试。最后改用手动TTL+死信交换机的方案：

```java
// 声明主队列
Map args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx.exchange");
args.put("x-dead-letter-routing-key", "dlx.routing.key");
args.put("x-message-ttl", 5601000); // 5分钟TTL
channel.queueDeclare("business.queue", true, false, false, args);

// 声明重试队列
Map retryArgs = new HashMap<>();
retryArgs.put("x-dead-letter-exchange", "");
retryArgs.put("x-dead-letter-routing-key", "business.queue");
retryArgs.put("x-message-ttl", 5601000);
channel.queueDeclare("retry.queue", true, false, false, retryArgs);
```

消费者处理失败后，不直接重试，而是把消息发到死信交换机，等TTL过期后自动回到原队列。每重试一次，消息头里记录重试次数，超过3次就转发到真正的死信队列。说实话这个方案比插件稳定多了，但是调试的时候需要特别注意死信交换机的路由规则，我一开始箭头指错了，消息直接石沉大海。

最后总结一下，RabbitMQ的可靠性投递核心就三件事：生产端用confirm机制、消费端用幂等、重试用TTL配合死信。不要迷信复杂方案，数据库唯一索引是最靠谱的幂等方案。本文基于实际项目经验整理，如有技术问题欢迎私信交流。