消息队列-消息重复

by zhuzhenPosted on

生产者重复发送：生产者发送消息后，因为网络超时等原因没收到 Broker 的确认，它无法判断消息是否发送成功，为了保证消息不丢失，通常会选择重试。这就可能导致同一条消息被发送了多次

消费者侧的重复消费：消费者拉取消息，业务逻辑处理完了，正准备提交消费位点（ACK）时，服务突然宕机或重启。当服务恢复后，它会从未提交的位点重新拉取消息，导致同一条消息被再次消费。

以Kafka为例，消费者的工作流程是：拉取消息 → 处理消息 → 提交offset（告诉Broker”我已经消费到第N条了”）。关键问题在于”处理消息”和”提交offset”是两个独立的步骤，不是原子操作。假设消费者拉了第100条消息，处理完了，正准备提交offset=100，这一瞬间消费者宕机了。重启后Broker发现上次提交的offset还是99，于是又把第100条消息推给消费者——重复消费就这么发生了。

2.1 数据库唯一约束

这是最简单、最直接，也是最常用的一种方案。其核心思想是，利用数据库中“唯一索引”或“主键”的特性，来阻挡重复数据的插入。

假设我们有一个电商系统，用户下单后会发送一条消息，触发给用户增加积分的操作。消息内容可能包含{ "order_id": "202508310001", "user_id": 58, "points_to_add": 100 }。

这个“增加积分”的操作，天然是非幂等的。我们可以这样改造：

建立一张积分流水表（points_log）。表中包含字段：id (自增主键), order_id (订单ID), user_id (用户ID), points (变更积分), create_time。关键一步： 对 order_id 这个字段建立一个唯一索引。

    -- 尝试插入积分流水记录
-- 假设 order_id 字段上有唯一索引
INSERT INTO points_log (order_id, user_id, points) VALUES ('202508310001', 58, 100);

    第一次消费：该订单ID首次出现，INSERT操作成功。然后我们可以安全地去更新用户的总积分。

    重复消费：MQ再次投递相同的消息，消费者尝试INSERT时，数据库会因为order_id的唯一索引冲突而直接报错。我们的代码捕获这个异常后，就可以知道这是重复操作，直接忽略并返回ACK即可。

    这种方案的优点是： 实现简单，成本低，效果可靠。 缺点也很明显： 强依赖数据库特性，对于非数据库操作的场景无能为力。基于这个思路，如果不用关系型数据库，Redis的SETNX命令（SET if Not eXists）也能达到异曲同工的效果，可以用order_id作为key，实现分布式锁或状态记录。

    2.2 版本号机制

    上面的方案都是基于数据插入场景的，假设我们的业务操作不是数据插入，而是数据更新呢

    确实，如果我们的业务不是INSERT，而是UPDATE呢？比如，更新订单状态。这时，唯一约束就派不上用场了。我们可以引入“前置条件”或“版本号”机制，也就是常说的乐观锁。

    假设有这样一个场景，订单支付成功后，需要将订单状态从“待支付”（status=1）更新为“待发货”（status=2）。消息内容为{ "order_id": "202508310002", "target_status": 2 }。

    直接执行UPDATE orders SET status = 2 WHERE order_id = '202508310002'是非幂等的。如果因为某种原因，后续还有一个“取消订单”的操作把状态改回了1，这条重复的消息可能会错误地再次把订单改为“待发货”。

    我们可以这样改造：

    在orders表中增加一个version字段，默认为0或1。消费消息时，我们从消息中（或者先查询一次数据库）拿到当前的版本号。执行UPDATE时，带上版本号作为条件。

    -- 更新订单状态，同时检查版本号
-- 假设当前数据库中该订单的 version 是 5
UPDATE orders 
SET status = 2, version = version + 1 
WHERE order_id = '202508310002' AND version = 5;

    第一次消费：version为5，条件满足，UPDATE成功。数据库中的version变为6。

    重复消费：MQ再次投递消息，消费者执行同样的SQL，但此时数据库中的version已经是6了，不满足AND version = 5的条件。UPDATE语句会执行失败，影响行数为0。我们就知道这是重复操作了。

    同样在分析完这个方案之后，你可以做一个方案优缺点的补充。优点： 适用范围比唯一约束更广，能处理大部分更新操作。 缺点： 需要在业务表中增加额外字段（如version），有一定侵入性。

    防重表

    如果我们的业务逻辑非常复杂，可能涉及多个表的更新，甚至是一些外部RPC调用，这个时候版本号已经不起作用了，此时应该怎么办呢？

    这个时候就到了我们祭出我们第一个亮点方案的时候了：全局唯一ID + 单独的防重表（或缓存）

    防重表也叫幂等记录表，这个方案的核心思想是，为每一次消息处理操作生成一个全局唯一的标识。在执行核心业务逻辑前，先将这个唯一标识插入一张“幂等记录表”或直接利用业务表中的唯一约束字段。如果插入成功，说明是首次处理，继续执行业务；如果插入失败（因为唯一键冲突），则说明这条消息已经被处理过了，直接丢弃即可。

    在面试的时候你可以先介绍下这个方案的基本流程

    如果业务复杂，我们可以采用防重表的方案，将业务逻辑和幂等逻辑解耦。单独建立一张防重表，具体的步骤如下：

    为每条消息生成一个全局唯一ID（GUID）。这个ID可以在生产者发送时就放入消息体或Header中。

    建立一张“消费记录表”（consumed_log），表结构很简单，核心就是一个字段message_id，并将其设为主键或唯一索引。

    消费者处理逻辑变为一个“三段式”：

    a. 开启事务。b. INSERT消息的GUID到consumed_log表中。c. 执行真正的业务逻辑（更新数据库、调用RPC等）。d. 提交事务。

        这样如果是重复消息的话，就会插入消费记录表失败，就不会执行后面的业务逻辑了

        异步校对

        上面提到了用事务，在一个数据库里确实没有问题，可以用本地事务来保证防重逻辑和业务逻辑的原子性，但是如果是分布式环境下，跨库要怎么处理呢？

        这里如果你能把分布式环境下的跨库幂等性实现也讲清楚的话，其实就已经可以跟一般候选人拉开差距了。确实，在微服务架构下，业务操作往往是跨服务的，比如“扣减库存”和“创建物流单”可能分别由两个不同的服务实现。这时，本地事务就失效了。

        此时，我们需要引入最终一致性的设计思想，并辅以一个异步校对机制。整个流程会演变成三步：

        1：预操作：收到消息后，第一步是在幂等记录表中插入一条记录，但状态标记为“处理中（PROCESSING）”。例如，插入一条记录 (order_id, 'PROCESSING')。这一步是幂等性的关键防线。2：执行业务：调用库存、物流等下游服务，执行核心业务逻辑。3：确认操作：所有业务逻辑成功执行后，回来将幂等记录表中的状态更新为“已完成（COMPLETED）”。

          这里由于没有事务保证，所以很可能出现第二步执行业务成功了，但第三步更新幂等表对应数据为已完成的时候失败了（比如网络问题或服务宕机）。这时，幂等记录表里会留下一条“处理中”的“记录。

          这个时候就是异步校对机制发挥作用的时候了。它会定期扫描幂等记录表中那些长时间处于“处理中”状态的记录，然后反向查询各个业务系统（比如查询物流系统是否存在该订单的物流单），来判断业务是否真的执行成功。如果查询下来业务确实已经成功，校对任务就负责将幂等记录的状态更新为“已完成”。

          “如果查询下来业务也没有成功，会怎么样呢？”

          其实这个时候还是一致性的状态，就说明业务确实是没有执行成功，所以不会修改状态为已完成，这个时候可以直接重新再出发一次业务操作就可以了，可以设置一个重试次数，如果超过重试阈值，一致不成功，最后只能由人工介入

          用redis缓存判端消息重复

          但所有请求，无论是首次还是重复的，最终都要访问数据库。在每秒几万甚至几十万请求的场景下，数据库很快就会成为瓶颈。有什么优化思路？

          当一个新的消息进来的时候，我们先通过redis做一次判重校验，如果这个key存在，那么我们就认为这是重复的key，如果redis不存在，再通过数据库做一次兜底校验，如果key存在就认为是重复的消息，如果key不存在，就认为不是重复消息，没处理过

          一旦引入缓存，就涉及到缓存和数据库的一致性问题了，那这里我们每次处理一个新的消息之后，是先更新数据库（防重表）呢，还是先更新缓存redis呢？

          这里一定是先更新数据库，因为它是最可靠的，也是我们的兜底方案，你可以这样回答

          处理完业务逻辑之后，先更新数据库，把这个新的消息写入到防重表，在更新redis。这里即使redis更新失败，也没有关系，下一次这个重复的消息过来的时候，做重复性校验的时候，无非就是redis这里的拦截不起作用了，但是还会透穿到数据库层面去做校验， 还是能把重复消息拦截掉。保证消息幂等消费

          那这里如何确定redis里key的过期时间呢？

          关于redis的key过期时间其实没有一个严格的标准，一般是根据业务场景来定的，可以先观察具体消息队列接入的业务key的出现频率来设定，通过测试观察比如key的过期时间大概是5分钟出现一次的话，那我们比过期时间稍微设置长一点时间即可，比如六七分钟都是可以的。但是这里的key过期时间不宜设置过长，比如半小时，一小时。如果设置的太长，如果业务量大，并发量高的话，会造成redis的存储量暴涨，引起redis瓶颈

          布隆过滤器

          如果数据库是瓶颈，那么优化的核心思路就是：尽可能地减少对数据库的直接访问。redis缓存已访问的消息，又达不到性能要求，那我们可以就在数据库前构建一个或多个高性能的过滤层，让绝大多数的重复请求在到达数据库之前就被拦截掉。

          思考一下，这里我们要的是什么，没有出现过的直接放行，出现过的拦截掉。这不正是布隆过滤器的用武之地吗。所以我们可以在数据库前面加一层布隆过滤器作为拦截层。