帮客户落地RabbitMQ3个可靠性投递与重复消费实战要点前不久帮一个做招聘SaaS的客户搞了个项目他们的核心业务是职位信息同步。客户那边有几个数据源每天要定时从不同的招聘网站把最新的职位数据同步到自己的平台上同时还要把平台上的职位状态变更推送给外部系统。原来的方案是用数据库轮询Kafka但运维成本和网络开销扛不住了特别是网络抖动频繁导致数据不一致。客户希望引入RabbitMQ来解耦要求消息不能丢而且不能重复消费。项目规模大概是每天处理50万条消息峰值3000条/秒左右。可靠性投递生产端确认机制一开始我觉得RabbitMQ设个手动ACK就完事了结果发现生产端才是第一道鬼门关。测试的时候模拟了断网、Broker宕机发现消息在发送过程中就丢了消费者那边连个影子都没见到。当时有两种选择事务机制和publisher confirm机制。RabbitMQ的事务确实能保证不丢消息但性能损耗太大了实测TPS直接掉到原来的三分之一还多。我最后选了publisher confirm异步回调的方案虽然代码复杂度上去了但性能和可靠性都能兼顾。核心代码大概长这样java// 开启确认模式channel.confirmSelect();// 发送消息的回调channel.addConfirmListener(new ConfirmListener() {Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {// 确认成功从本地缓冲中移除if (multiple) {// 批量确认messageBuffer.headMap(deliveryTag 1).clear();} else {messageBuffer.remove(deliveryTag);}}Overridepublic 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 statusreturn 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分钟TTLchannel.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配合死信。不要迷信复杂方案数据库唯一索引是最靠谱的幂等方案。本文基于实际项目经验整理如有技术问题欢迎私信交流。