很多人一提到消息队列重复消费第一反应就是“加个幂等”但真到动手写方案的时候往往会发现里面全是细节什么时候会产生重复手动确认和自动确认差在哪唯一ID到底用谁的Redis去重和数据库唯一索引怎么选条件竞争怎么处理这些问题不搞清楚代码写出来心里没底线上出问题也兜不住。我早期做订单消息处理的时候就因为重复投递导致同一笔订单被发送了两次发货通知客户那边连收两条短信差点被投诉。后来把整套解决方案梳理了一遍才彻底把这个坑填平。这篇文章把我踩过的坑、验证过的方案、以及调研常见部署问题时顺便收集到的经验一起整理出来希望能给正在搞RabbitMQ的同行省点时间。1. 先从根上搞清楚RabbitMQ为什么会产生重复消费1.1 投递语义决定你躲不开重复RabbitMQ默认的消息投递模式是at-least-once也就是至少一次。这个语义的意思是消息不丢但可能重复。很多人在学MQ的时候只记住了“不丢失”把“至少一次”里的“至少”两个字给漏掉了这就为后面埋下了一颗雷。为什么会这样因为MQ为了保证“不丢”必须在收到消费者的确认信号后才把消息从队列里删掉。如果消费者处理完消息、但在发送确认之前崩溃了或者网络连接断了Broker收不到确认就会认为这条消息没被处理完于是重新投递给其他消费者。这是分布式系统里“为了不丢而不得不容忍重复”的典型取舍。如果这时消费者端没有做任何保护同一条业务消息就会被执行两次。对于发短信、扣库存、转账这种操作重复执行的后果是很严重的。所以只要用的是RabbitMQ默认的确认模式重复消费就不是“会不会发生”的问题而是“什么时候发生”的问题。1.2 重复消费的四个典型来源我总结了实际项目中最容易触发重复投递的四个场景发生频率从高到低排列如下消费者处理超时Broker重新投递默认情况下RabbitMQ的消费者如果长时间不发送ackBroker会在连接超时后认定消费者失联然后把消息重新入队。这个超时时间受心跳参数heartbeat timeout影响。如果业务处理耗时大于心跳间隔且代码里没有及时刷新心跳就特别容易出现重复投递。消费者逻辑执行完毕但ack失败这是最隐蔽的一种。业务代码已经处理完消息但在调用channel.basicAck()时连接断了Broker以为消息还没处理重新投递。你代码里的业务逻辑已经执行过一次现在又要执行第二次。生产者发送时重试导致重复写入生产端调用channel.basicPublish()时如果连接异常很多生产端代码会做retry。有些retry处理得粗糙没有判断上一次的publish其实已经成功导致同一条消息被发送了两次。消费者端在autoAck模式下宕机autoAck是在消息被交给消费者回调函数的一瞬间就自动确认。如果消费逻辑执行到一半进程崩溃这条消息就丢了重启后会从队列里再取一遍吗不会它已经确认了。所以autoAck模式下一般不丢但如果框架在确认后又拉到了新的消息而业务处理失败了还是会在业务层面出现“重复执行”的错觉。这四个来源分开看都不难理解但现实中往往是叠加发生的。所以你在设计幂等方案时不能只防一种情况要假设最坏情况消息可能被多次投递投递顺序可能变化消费进程可能随时崩溃。1.3 自动确认和手动确认到底差在哪RabbitMQ消费者有两种确认方式。自动确认autoAcktrue是只要消息被投递给消费者回调MQ立刻把它标记为已消费不管业务代码是否处理成功。手动确认autoAckfalse则是等你业务处理完主动调用ack告诉Broker处理成功了Broker才删除消息。从重复消费的角度看自动确认的问题是如果业务处理失败消息已经确认了想重试都不知道去哪找原消息。手动确认的问题是消息不自动确认如果你在ack前崩溃Broker会重新投递从而产生重复消费。注意手动确认是“不丢消息”的前提但也是“重复消费”的诱因之一。两者是同一个机制的两面不能只要一边。那是不是说用自动确认就可以避免重复了也不是。自动确认虽然避免了“ack前崩溃导致重投”的路但因为消费者进程可能在业务逻辑执行到一半时崩溃进程重启后的恢复机制、或者业务代码本身的重试逻辑照样可能把同样的事件再触发一次。所以不管你选哪种确认模式幂等防线都必须在业务侧建起来。2. 化解重复消费的核心思路消费端幂等2.1 用生活化类比理解幂等幂等Idempotency这个词听起来有点学术但背后的道理很朴素同一个操作执行一次和执行一百次最终结果必须是一样的。举个生活化的例子你坐电梯按了10楼按钮按第一次电梯动了。如果因为没看清又按了一次电梯不会从10楼变成20楼按钮按几次都是去10楼这就是幂等。反过来如果你在便利店买了两次同样的商品扣了两次钱那就不是幂等。对应到消息处理场景幂等的目标就是同一条消息无论被RabbitMQ投递多少次最终对业务数据产生的影响只能有一次。2.2 为什么不在Broker层做全局去重很多人会问RabbitMQ能不能像Kafka那样直接靠日志offset天然避免重复或者能不能在Broker端直接去掉重复消息先解释一下为什么不能。RabbitMQ的队列模型是消息被消费并确认后就删除Broker没有保留所有历史消息的全局日志。它没有地方去维护“哪些消息已经被哪些消费者处理过”的状态也无法区分两条内容相同的消息到底是“同一条重投”还是“两条正常的相似消息”。让Broker去做业务级的去重等于让框架背业务的锅设计上就是错的。所以重复消费的最终解决方案必须下沉到消费者也就是所谓的“消费端幂等”。这是目前行业里的统一解法有且只有一个核心原则把业务数据和消息的唯一标识绑定让重复投递变成无害操作。2.3 幂等方案的三层设计我在实际项目里把幂等方案的形态分成三层你可以根据自身业务选择合适的组合存储层约束利用数据库唯一索引在物理层面杜绝同一条消息被插入两次。这是最硬的一层防线可靠性最高。缓存层标记用Redis的SETNX命令或者记录处理状态在逻辑层面做快速过滤。适合抗并发要求高、且容忍极小概率脏数据的场景。业务层校验在业务流程中校验当前状态是否允许本次操作比如订单已经支付了就不能再次支付。这一层专门用来处理“状态变更型”业务。三层方案不是互斥的而是层层递进的关系。我最终在项目中采用的组合是消息唯一ID 数据库唯一索引 业务状态机校验。这套组合可以覆盖绝大多数场景。3. 实操三种落地方案的细节对比与选型建议3.1 方案一消息唯一ID 数据库唯一约束这是我最推荐的基础方案。核心思路生产者在发送每条消息时为消息生成一个全局唯一的业务ID比如订单ID 事件类型 事件序号消费者收到消息后先把这个ID插入到一张独立的“消息处理记录表”中。如果插入成功说明这条消息是第一次处理继续执行如果插入时因重复键冲突失败说明之前处理过直接丢弃。这里有个容易被忽略的关键点消息ID必须携带业务语义不能直接用全局UUID。假设某个订单ID是202501281000001这个订单发生了“创建”事件和“支付”事件它们对应的消息ID就应该是ORDER.CREATE.202501281000001ORDER.PAY.202501281000001这样做的好处是即使生产端因为网络重试把同一事件的同一条消息发了两次它携带的ID也是相同的消费者靠这个ID才能认出“这是同一件事”。如果你用UUID代替每条消息生成的UUID都不一样重投的消息也带上新UUID幂等就形同虚设了。数据库表设计也比较固定CREATE TABLE mq_message_record ( id BIGINT PRIMARY KEY AUTO_INCREMENT, msg_id VARCHAR(64) NOT NULL COMMENT 业务唯一消息ID, msg_type VARCHAR(32) NOT NULL COMMENT 消息类型, consume_time DATETIME NOT NULL COMMENT 消费时间, status TINYINT NOT NULL DEFAULT 0 COMMENT 0-处理中 1-成功 2-失败, UNIQUE KEY uk_msg_id (msg_id) ) COMMENT 消息消费记录表用于幂等去重;消费者核心逻辑就三步用INSERT IGNORE或SELECT ... FOR UPDATE判断消息是否已经处理过。如果数据库中不存在执行真正的业务逻辑。业务逻辑执行成功后提交事务这里要特别注意事务范围下面会细讲。这个方案的优势是物理约束绝不漏判适合金融、订单等强一致场景。劣势是每次消费都要多一次数据库写入吞吐量上会有损耗。3.2 方案二Redis SETNX 快速去重如果业务量比较大、追求处理性能可以把“是否处理过”的判断从数据库挪到Redis。核心操作是SET key value NX EX timeout只有key不存在时才能设置成功。用消息ID作为key如果设置成功说明第一次处理如果设置失败说明之前已经处理过了直接ack并跳过业务逻辑。举个例子SET mq:order:pay:202501281000001 1 NX EX 86400这条命令的意思是如果key不存在就设置成功并设置一天过期时间如果key已经存在则不做任何操作。在Java里面的写法大致是这样的boolean firstConsume redisTemplate.opsForValue() .setIfAbsent(mq: msgId, 1, Duration.ofHours(24)); if (!firstConsume) { // 消息已经处理过直接返回 return; } // 处理真正的业务逻辑这个方案我在压测环境里跑过qps能比数据库方案高出一个数量级。但它的短板也很明显Redis的过期时间如果设置得太短消息重投的间隔超过过期时间就会漏判如果设置得太长又会堆积大量无用key占据内存。另外如果你的Redis不是强一致集群极端情况下也有一点误判概率。所以我的建议是Redis方案适合对一致性要求没那么苛刻的场景比如通知类、日志类、统计类核心交易数据最好还是落在数据库约束上。注意Redis的过期时间建议设置成消息最大重投间隔的5倍以上。如果你不确定重投间隔上限就设置7天不要设太短。3.3 方案三状态机校验这一层解决的是“同一条业务事件不该反复触发状态流转”的问题。典型场景订单已支付就不该再支付一次退款单已关闭就不该再关闭一次。实现方式是在业务表上维护一个状态字段在更新数据前先按当前状态做一次条件判断。比如int rows orderMapper.updateStatusIfMatched( orderId, expectedStatus, // 期望当前状态 targetStatus // 目标状态 ); if (rows 0) { // 状态不匹配说明已经有别的消息处理过了或者消息本身就是旧的 return; }这里的关键SQL是带WHERE status expectedStatus条件利用数据库行锁保证并发安全。如果更新的影响行数是0就说明当前状态不是预期的状态直接放弃处理不再继续后面的流程。这个方案单独用其实挡不住“同一状态重复执行两次”的问题因为某些操作在相同状态下也可能是合法的。所以它的定位是辅助性防线在存储层约束和缓存层标记之后再加一道业务逻辑校验让整套幂等更立体。3.4 方案选型速查说了三个方案很多人会纠结到底用哪个。我直接按照业务场景给你一个速查表场景推荐方案原因订单、支付、库存等核心交易唯一ID 数据库唯一约束强一致物理防重通知、短信、邮件等消息推送Redis SETNX高吞吐过期时间可控积分、优惠券等非资金类数据Redis SETNX 状态机性能与一致性平衡数据同步、日志采集唯一ID Redis允许极小概率丢失追求速度涉及多表变更的复杂事务唯一ID 数据库约束 本地事务表跨表场景必须有回滚保障这张表不是金科玉律但它能帮你避开“单号已创建就疯狂发短信”这种低级坑。选型的核心逻辑永远是业务对数据不一致的容忍度有多高决定了幂等方案的强度。4. 实操过程从生产端到消费端的完整落地4.1 环境准备RabbitMQ部署的几个前提先说一下部署层的准备工作因为很多团队在处理重复消费问题时连环境都不太稳。我整理了两个高频问题。第一个是Windows环境的安装问题。很多实训项目、本地调试都在Windows上进行而RabbitMQ依赖Erlang版本匹配是个坑。需要用官网提供的版本兼容对照表来选版本不要图省事直接装最新版。装完之后如果服务无法启动多半是Erlang路径没加载或者端口被占用优先检查环境变量ERLANG_HOME是否配置正确。第二个是端口修改问题。RabbitMQ默认监听5672端口如果和本机其他应用冲突可以在rabbitmq.conf中显式修改listeners.tcp.default 5673注意修改端口后所有客户端连接参数也要同步改否则会出现“连接被拒绝”的假象。还有一种情况是远程连接失败这不是端口的问题而是RabbitMQ默认只绑定本机回环地址。如果要在局域网内调试必须创建一个专用于远程访问的用户并赋予虚拟主机权限不能直接用guest默认账号连接远程IP。4.2 生产端发送消息时的唯一ID生成规则幂等的第一道关口在生产端。你在发送消息时不能随手生成一个随机UUID了事而是要在业务层面规定好消息ID的生成规则。我推荐的做法是这样的把“业务主键 事件版本号”拼成消息ID。举个例子在Spring Boot中使用RabbitTemplate发送消息String msgId order.getOrderNo() : eventType : eventVersion; MessageProperties props new MessageProperties(); props.setMessageId(msgId); Message message MessageBuilder .withBody(JSON.toJSONBytes(order)) .andProperties(props) .build(); rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message);这段代码里eventVersion的作用很重要。如果同一个订单的同一种事件因为业务原因需要重新发送比如支付成功通知发送失败后人工触达一次可以通过递增version来区分“同一次事件的不同版本”。这样幂等判断不仅限于“同一ID不重复处理”还顺带支持了“新版本可以继续处理”。4.3 消费端手动确认与幂等判断的配合下面给出一个我在项目中实测过的完整消费端代码骨架用的是Spring Boot RabbitMQ的默认配置确认模式设为手动spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest listener: simple: acknowledge-mode: manual prefetch: 10消息消费者的核心逻辑是RabbitListener(queues order.pay.queue) public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag message.getMessageProperties().getDeliveryTag(); String msgId message.getMessageProperties().getMessageId(); try { // 1. 基于业务唯一ID做幂等判断 boolean handled checkIfHandled(msgId); if (handled) { channel.basicAck(deliveryTag, false); return; } // 2. 执行真正的业务逻辑 OrderPayMessage payMessage JSON.parseObject( message.getBody(), OrderPayMessage.class); handleOrderPay(payMessage); // 3. 幂等记录与业务逻辑在同一个本地事务里 markHandled(msgId); // 4. 手动确认 channel.basicAck(deliveryTag, false); } catch (Exception e) { // 业务异常记录日志 log.error(消费消息失败, e); // 确认失败并重新入队 channel.basicNack(deliveryTag, false, true); } }这套代码关键点有三个。第一个是markHandled(msgId)必须和业务变更在同一个本地事务中执行否则可能出现“业务数据改成功了、但幂等记录没写进去”的中间态。第二个是checkIfHandled不能先查再写因为并发条件下两个消费者同时查到“未处理”就会同时执行业务逻辑。要在数据库层面用唯一索引约束做“写时校验”或者用分布式锁保护。第三个是basicNack的requeue参数要谨慎使用如果业务逻辑本身有问题且没做次数限制无限重投会让消息永远在队列里转圈。4.4 事务边界的正解业务操作与记录必须在同一个事务这一步是整个方案的灵魂也是新手最容易搞错的地方。很多人写代码的时候分别执行“业务操作”和“插入幂等记录”然后自信地以为没问题。实际上这两个操作缺了事务保护就像两个不相干的人分头行动中间任何一步崩溃都会造成结果不一致。正确姿势是始终把幂等校验、业务数据变更、消息处理记录三者放在同一个 Transactional 事务中Transactional(rollbackFor Exception.class) public void handleOrderPay(OrderPayMessage msg, String msgId) { // 1. 尝试插入幂等记录利用唯一索引防重 int inserted mqMessageRecordMapper.tryInsert(msgId); if (inserted 0) { throw new DuplicateMessageException(); } // 2. 更新订单状态 int updated orderMapper.updateStatusIfMatched( msg.getOrderNo(), UNPAID, PAID); if (updated 0) { throw new IllegalStateException(订单状态不允许跳转); } // 3. 写其他业务表 // 全部成功后事务提交 }这样设计的逻辑是如果业务数据变更成功幂等记录也在同一事务里提交如果后续某一步失败回滚幂等记录也会跟着回滚这条消息还能被重新投递时再处理一遍。整条链路不会出现“改了数据但没记录”或“记录了但没改数据”的分裂状态。4.5 并发场景下的幂等验证方法写完方案后千万不要在单线程环境里自嗨。重复消费的高危场景是并发投递——比如多个消费者实例同时消费了同一队列的同一消息。我常用的验证工具是JMeter或直接写一个多线程测试类模拟多个线程同时消费同一条消息。步骤大致如下准备一个带唯一订单号的测试消息producer发送一条。在消费逻辑里故意加入CountDownLatch让所有消费线程同时开始处理。观察数据库中的订单状态是否为期望值幂等表中的记录只有一条。用JMeter并发曲线验证高tps下不会出现重复扣款、重复加积分这类问题。我在压测中确实遇到过这样的场景数据库唯一索引没建好时两个线程同时插入同一条消息因为表里没唯一约束两条都插进去了。加上唯一索引后MySQL的DuplicateKey报错帮我们挡住了第二个线程的写入。这一测试让我彻底明白代码里的if判断只是逻辑上的防御真正可靠的兜底是数据库的物理约束。5. 常见问题与排查技巧实录5.1 消息被重复投递但业务逻辑幂等了还需要做什么很多人完成幂等方案后觉得万事大吉其实还有两个细节要处理。一是消息积压问题如果业务逻辑比较耗时prefetch值太小会导致同一批次的消息处理不过来看起来就跟重复消费一样买基金。二是重试次数上限避免“毒消息”无限重投。可以在消息里加一个retryCount字段每次被消费时加1超过3次就转到死信队列由人工介入处理。5.2 手动确认模式下为什么日志里还是看到了重复消费这是排查频率最高的问题。代码里确实调用了basicAck日志却显示同一条消息执行了两次。我遇到过两种原因。第一种是业务逻辑耗时超过了RabbitMQ的心跳超时时间连接被判定失效Broker在连接关闭后把未确认的消息重新入队。这时候要调整心跳超时和连接超时的配置增大消费者实例的线程池并且建议把耗时操作异步化或者拆分多条消息。第二种是幂等记录表和业务表不在同一个数据库。比如业务数据在MySQL幂等记录放在了RedisMySQL事务提交成功后Redis写入却失败了。这样消息被重新投递时幂等记录里查不到自然就会再次执行业务逻辑。解决办法是强一致场景下幂等记录一定丢进和业务数据同一个事务。5.3 消费失败后basicNack、basicReject、死信队列怎么选很多新人对RabbitMQ消费失败的三个方法区别不清我直接列个对比方法效果适用场景basicAck(deliveryTag, false)确认处理成功业务执行成功或业务判定重复消息basicNack(deliveryTag, false, true)拒绝并重新入队临时故障但可能造成无限重投basicNack(deliveryTag, false, false)拒绝且不重新入队消息不可处理需转入死信队列basicReject(deliveryTag, false)拒绝单条消息与basicNack类似但不支持批量拒绝我在实践中会尽量避免无限制的requeue而是给消息附带最大重试次数。达到上限后采用manual-negative-acknowledgment-topology-queue思路把消息转发到专门的死信交换机DLX和死信队列。运维侧可以写个定时任务扫描死信队列人工或自动化重启处理流程。提示配置死信队列时需要给原队列绑定x-dead-letter-exchange和x-dead-letter-routing-key参数。消息被Nack且不requeue后会自动进入指定的死信队列不会直接丢掉。5.4 RabbitMQ环境部署的常见坑安装、启动与端口问题在排查重复消费问题时我发现很多团队其实环境就没搭稳。有几个高频问题顺带列一下方便大家自查Windows安装RabbitMQ报错服务无法启动优先检查Erlang版本是否匹配配置ERLANG_HOME环境变量。RabbitMQ启动失败日志显示clean channel shutdown通常是磁盘空间不足或内存水位告警触发RabbitMQ在内存或磁盘超过阈值后会主动阻塞生产者并关闭连接。这时候检查rabbitmq-diagnostics memory和rabbitmq-diagnostics disk_free把水位调低或者扩容。修改端口后客户端无法连接改完端口要重启服务并同步修改客户端连接参数。另外远程访问需要设置NODE_PORT环境变量而不是只改配置文件。Linux下apt或yum安装旧版本建议使用RabbitMQ官网提供的独立包或Docker镜像避免系统自带源里的版本过于陈旧。这些问题的排查思路其实都一个套路先看日志再查资源和端口最后看版本兼容。不要一上来就重装系统浪费时间。5.5 如何确认你的幂等方案真的生效了最后分享一个我觉得特别实用的验证手段在生产环境的灰度阶段故意制造一次模拟重复投递。具体做法是写一个测试接口调用生产者的重发逻辑把最近一条业务消息再发一次。然后观察消费端日志和数据库状态确认第二条重复消息被幂等拦截、业务数据没有被二次变更。我几乎在每个项目上线前都会做这步验证。别小看这个动作它能一次性检验生产者的重试逻辑、消费者的幂等判断、数据库事务边界是否同时达到要求。如果这一关过了线上再出现意外重投你才能睡个安稳觉。我个人在实际项目中最深的体会是消息中间件本身不解决重复消费它只保证“不丢”重复的问题必须由业务系统自己兜底。与其在代码里到处加判断、企图拦截所有可能的重复分支不如设计一个清晰的消息ID规范配合一个存储层的唯一约束把整件事从“逻辑防重”升级到“物理防重”。这套组合拳打下来才是真正让人放心的方案。
