1.如何防止重复消费以rbmq为例其它原理都一样消费者拿到消息 → 开始处理处理太慢超时了RabbitMQ 自动认为消费失败 →隐式 NACK队列把这条消息重新放回队列再发一次这时候另一个消费者或同一个消费者的另一个线程又收到这条消息开始执行第二遍关键来了原来那个慢的消费者根本不知道超时了还在继续跑最后老线程执行完了新线程也执行完了→同一个业务逻辑执行了两遍 → 重复消费把消息给消费者放到消费者内存里但消息本体还在 MQ 队列里一直都在只是状态变成unacked被拿走待确认等待结果ACK→ 队列里这条消息删掉NACK / 超时→ 状态变回ready再发一遍下面来看解决方法分布式锁 状态幂等 手动 ACK/NACK 重试 3 次进死信 乐观锁 防重复消费下面来看代码spring: rabbitmq: host: localhost port: 5672 username: guest password: guest listener: simple: # 必须手动ACK超时才生效 acknowledge-mode: manual # 消费超时时间毫秒 consumer-timeout: 60000 # 60秒消费者完整代码import com.rabbitmq.client.Channel; import lombok.RequiredArgsConstructor; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.concurrent.TimeUnit; Component RequiredArgsConstructor public class OrderConsumer { private final StringRedisTemplate redisTemplate; private final OrderMapper orderMapper; private final RabbitTemplate rabbitTemplate; // 监听业务队列 RabbitListener(queues order.queue) public void handleOrderMessage(String orderId, Message message, Channel channel) throws IOException { // 消息唯一标识必须用这个ACK/NACK long deliveryTag message.getMessageProperties().getDeliveryTag(); // 分布式锁key用订单id做唯一标识 String lockKey lock:order: orderId; // 从消息头获取已重试次数默认0 Integer retryCount (Integer) message.getMessageProperties() .getHeaders().getOrDefault(retry-count, 0); try { // 1. 先加Redis分布式锁 // 30秒过期防止服务宕机死锁 Boolean lockSuccess redisTemplate.opsForValue() .setIfAbsent(lockKey, 1, 30, TimeUnit.SECONDS); // 拿不到锁说明有线程正在处理直接拒绝不重试 if (Boolean.FALSE.equals(lockSuccess)) { channel.basicNack(deliveryTag, false, false); return; } // 2. 幂等判断 Order order orderMapper.selectById(orderId); // 订单不存在/不是待支付状态说明已经处理过直接ACK不重复执行 if (order null || !WAIT_PAY.equals(order.getStatus())) { channel.basicAck(deliveryTag, false); return; } // 3. 乐观锁CAS更新核心兜底 // 只有状态是WAIT_PAY才能改成PROCESSING原子操作 int updateRows orderMapper.casUpdateStatus(orderId, WAIT_PAY, PROCESSING); // 更新行数0说明被其他线程抢先处理了直接ACK结束 if (updateRows 0) { channel.basicAck(deliveryTag, false); return; } // 4. 真正执行业务逻辑 // 这里写你的核心业务扣库存、调用支付、生成物流单等 // ...... // 业务执行成功更新订单最终状态 orderMapper.casUpdateStatus(orderId, PROCESSING, SUCCESS); // 业务成功手动ACK消息从队列删除 channel.basicAck(deliveryTag, false); } catch (Exception e) { // 异常处理重试3次超过进死信 retryCount retryCount 1; if (retryCount 3) { // 重试满3次直接NACK不重回队列自动进死信 channel.basicNack(deliveryTag, false, false); } else { // 不足3次把重试次数塞进消息头重新投递到原队列 message.getMessageProperties().setHeader(retry-count, retryCount); rabbitTemplate.send(order.exchange, order.routingKey, message); // 原消息ACK避免重复 channel.basicAck(deliveryTag, false); } } finally { // 无论成功失败必须释放锁 redisTemplate.delete(lockKey); } } }OrderMapper 里的 CAS 方法Update(UPDATE order SET status #{targetStatus} WHERE id #{orderId} AND status #{expectStatus}) int casUpdateStatus(String orderId, String expectStatus, String targetStatus); Update(UPDATE order SET status #{status} WHERE id #{orderId}) int updateStatus(String orderId, String status);import com.baomidou.mybatisplus.core.mapper.BaseMapper; import org.apache.ibatis.annotations.Update; import org.springframework.stereotype.Repository; Repository public interface OrderMapper extends BaseMapperOrder { /** * 核心用status做乐观锁的CAS原子更新 * 只有订单当前状态是expectStatus才会更新成targetStatus * 数据库行锁保证同一时间只有1个线程能更新成功 */ Update(UPDATE order SET status #{targetStatus} WHERE id #{orderId} AND status #{expectStatus}) int casUpdateStatus(String orderId, String expectStatus, String targetStatus); }消息来了先加 Redis 分布式锁同一订单同时只能有一个线程处理防止并发重复。拿锁后先查数据库状态如果已经不是待支付说明处理过直接 ACK不再执行。执行业务并更新订单状态状态一旦改变天然实现幂等后面再来多少条消息都不会重复执行。然后在进行CAS判断如果说状态是WALL_PAY(待支付)就可以改然后改成已支付然后下面在判断是不是Processing是的话就改成已支付成功就 ACK消息删除告诉 RabbitMQ 消费完成。失败就计数最多重试 3 次超过 3 次直接 NACK 进死信不卡死队列。finally 里一定删锁避免死锁保证下一次能正常消费。用分布式锁防并发用数据库状态做幂等还有CAS用手动 ACK 保证可靠性用重试 死信避免死循环从根源彻底防止消息重复消费。为什么要加锁其实就是当第一个线程超时的时候然后发送NACK另一个线程也来执行逻辑所有判断都一路无阻然后在执行扣减库存然后就会多扣加了锁此时第一个线程拿到锁但是锁还没有过期即使他超时了然后其他消费者线程也执行这个逻辑是会被阻拦到拿锁那一步的。为什么要加幂等性就是当第一个消费者线程锁过期了然后第二个消费者线程进来然后如果说不加幂等判断当线程1执行完然后线程2也会继续执行然后就多扣了你进行幂等判断当线程1执行完那个status就会被改变了然后线程2就会被阻拦到这一步。为什么要用CAS就是当有一种情况就是当锁达到过期时间进行释放此时也超时了那么就会发送NACK然后就会再开一个线程执行这些逻辑然后那个线程拿到锁然后判断什么的都成功了此时原来那个超时的线程也要开始该状态了如果你不加cas那他俩都会进行改状态也会改成功然后当然下面的扣库存也会进行成功就没有办法防止重复消费了所以要加CAS。一、整套防重复机制总结用分布式锁防并发用数据库状态做幂等用 CAS 做最终兜底用手动 ACK 保证消息可靠性用重试 死信避免死循环从根源彻底防止消息重复消费。二、为什么要加分布式锁当第一个消费者线程执行超时MQ 会自动返回 NACK消息重新入队第二个线程就会开始执行。如果不加锁两个线程会同时一路执行到底都去扣库存、改状态导致重复扣减、数据错误。加了分布式锁后同一时间只有一个线程能拿到锁。即使第一个线程超时了只要锁没过期其他线程在拿锁这一步就会被直接拦住根本进不来从源头挡住并发重复。三、为什么要加幂等判断状态判断极端场景第一个线程的锁过期自动释放了第二个线程成功拿到锁。如果不加幂等判断当线程 1 执行完业务、线程 2 也会继续执行依然会重复扣库存、重复处理。加了幂等判断线程 1 执行完后订单状态会被修改。线程 2 进来后第一步就查到状态已变直接被拦截不再执行业务避免重复执行。四、为什么必须加CAS乐观锁最极端的场景第一个线程锁过期释放了同时消费超时返回 NACK。第二个线程进来拿到锁 状态判断都通过。此时原来的旧线程恢复了也要开始改状态、扣库存。如果不加 CAS两个线程都会修改成功都会扣库存重复消费无法避免加了 CASUPDATE order SET status ? WHERE id ? AND status WAIT_PAY这是数据库原子操作只有一个线程能更新成功。另一个线程更新行数 0直接被拦住绝对不会执行库存扣减。CAS 是最后一道防线彻底杜绝锁超时带来的重复执行问题。五、整套机制的三层防护最核心分布式锁挡住并发同一时间只让一个线程进状态幂等挡住已处理完的重复消息CAS 乐观锁兜底挡住锁过期的极端情况六、最终一句话最强总结分布式锁防并发状态幂等防重复CAS 兜底防锁超时三层防护层层兜底从根源彻底杜绝消息重复消费。RocketMQ如何保证消息的顺序性先来看为什么会出现顺序性问题主体A、B、C 是三个先后发起下单请求的用户正常业务要求必须先完整处理完先下单的 A 用户再处理 B再处理 C全局顺序不能乱问题根源 1生产者集群闯的祸你用了订单服务集群多实例部署A 用户的下单请求落到了实例 1B 用户的落到了实例 2C 用户的落到了实例 3。哪怕 A 用户是先点的下单、先触发的消息发送但实例 1 刚好遇到网络波动导致 A 用户的消息比后下单的 B 用户的消息晚到 MQ队列里的顺序直接变成了B → A → C入队顺序本身就错了。问题根源 2消费者集群补的刀你用了消费者集群多个实例同时监听这个队列。哪怕队列里顺序是对的A→B→CMQ 也会把 A 分给消费者 1B 分给消费者 2C 分给消费者 3并行处理。结果 A 的处理卡了B、C 先处理完了最终执行顺序变成了B→C→A哪怕入队顺序对执行顺序也乱了。要 100% 保证「全局用户请求严格按顺序处理」唯一的、缺一个条件都不行的方案就是你一直坚持的生产端禁用生产者集群只用单生产者单实例同一个线程严格按用户请求的先后顺序串行发送消息保证先发起的 A一定先入队彻底杜绝网络波动导致的入队乱序或者是用分布式锁它们共用同一把锁也就是锁都是长得一样的也就是那个key在生产者就拿发送成功后在释放然后再让别的去拿。如何处理RocketMQ消息积压问题一、源头控制生产者侧生产者限流限制消息发送速率避免生产速度远大于消费速度从源头减少队列压力。非核心消息降级 / 丢弃日志、统计、通知类非核心消息可临时关闭发送优先保障核心业务消息。二、消费速度优化消费者侧核心优化消费逻辑中的 SQL加索引、避免全表扫描、减少大事务、批量操作降低单条消息处理耗时。减少外部慢调用避免消费时调用慢接口、慢 RPC、冗余 IO能异步的都异步。提高消费并发增加消费者线程数 / 新增消费者实例提升整体消费能力。注意全局有序场景不能随便加并发调整预取数 prefetch适当调大消费者一次性拉取消息数量减少网络 IO 开销。关闭不必要的手动重试 / 死循环重试避免失败消息反复消费占用消费线程。三、队列架构急救严重积压时废弃原积压队列新建队列接收新流量旧队列不再使用新消息直接发到新队列保证当前业务不阻塞。旧队列消息转移到死信 / 临时队列存量积压消息转发到死信队列或临时队列用独立消费者慢慢消化不影响主流程。批量清理无效消息过期、重复、已失效的消息直接 ACK 丢弃快速减少队列长度。四、Broker 层面优化队列分片 / 多队列拆分按用户 ID / 订单 ID 哈希路由到多个队列并行消费提升吞吐量。清理队列冗余、避免队列过长提升 MQ 节点资源CPU、内存、磁盘五、兜底保障消费幂等防止重复处理导致业务异常死信队列兜底消费失败消息不丢失监控报警堆积数量、消费速度实时告警RocketMQ如何解决消息丢失问题1. 生产者发消息同步发送 事务注解1同步发送sync send生产者发送消息 →阻塞等待 Broker 确认收到SEND_OK才算发送成功优点可靠性最高消息不会丢缺点性能比异步、单向发送低适用场景订单创建、支付、扣费等核心业务消息绝不允许丢失。2加事务注解分布式事务消息以RocketMQ 事务消息为例流程生产者发送半消息 Half MessageBroker 响应半消息发送成功执行本地事务Transactional 注解根据本地事务结果向 Broker 提交COMMIT或ROLLBACK如果没提交 / 超时Broker 会回查生产者确认状态作用保证本地事务成功 ↔ 消息一定发送成功本地事务失败 ↔ 消息一定不发送彻底解决本地事务成功但消息没发出去 / 消息发了但本地事务回滚的问题。2. 异步发送失败回调机制处理异步发送producer.sendAsync(msg, callback)回调接口Callback两个方法onSuccess()消息发送成功onException()发送失败网络异常、Broker 宕机、流量超限等失败时做什么生产实战日志记录消息 ID、内容、时间、错误信息存入 DB 重试表定时任务重试重试 3~5 次仍失败 →告警 人工介入目的异步发送不能丢消息失败必须有兜底。3. 队列端同步刷盘什么是刷盘消息到 Broker 后从内存 → 写入磁盘的过程。同步刷盘SYNC_FLUSH消息写入磁盘成功后才返回ACK给生产者只要 Broker 回复成功消息一定落盘机器断电、宕机也不会丢异步刷盘ASYNC_FLUSH写到内存就返回成功后台线程批量刷盘性能高但断电可能丢少量消息你要记住一句话同步刷盘 数据绝对安全性能稍低异步刷盘 性能极高有极低丢消息风险金融、支付必须用同步刷盘。4. 消费者端消息重试机制最重点、最复杂1什么时候会重试消费者业务抛异常消费超时主动返回RECONSUME_LATER2重试规则RocketMQ 为例默认16 次重试间隔越来越长10s30s1min2min4min6min…直到最后 2 小时超过最大重试次数 → 进入死信队列DLQ3死信队列Dead Letter Queue重试多次仍失败的消息不会丢弃进入专门的死信队列人工排查数据问题、接口挂了、依赖异常等4消费重试必须注意消费者必须幂等重试会导致重复消费 → 必须保证重复消费不影响结果不要捕获异常后不处理异常吃掉 消息被认为消费成功丢消息慢消费会导致大量重试队列堵塞 → 系统雪崩5消费重试的本质Broker 收到消费失败响应 →不删除消息重新发送给消费者不是消息真的 “复制”是重投递。整体串一遍生产完整链路生产者同步发送保证消息不丢加事务注解保证本地事务与消息一致异步发送失败用回调记录 重试Broker同步刷盘消息落盘才确认消费者失败自动重试最终失败进入死信队列人工处理