Kafka 至少一次投递下，同一用户的事件偶发落到不同分区导致下游幂等失效，如何根治？

结论先行：你观察到的"旧 version 覆盖新值"几乎不可能是"跨分区重投"造成的，而是同一分区内消息重复投递 + 你的守卫只挡住了"旧值覆盖新值"但没挡住"同 version 二次写入" 叠加 消费端没有把 offset commit 与业务写入放进同一原子边界 的结果。真正的根治不在 Kafka 侧，而在下游把幂等键从 (userId) 升级为 (userId, version) 唯一键 + CAS，并让消费端做到"业务写入成功才提交 offset"。Kafka 侧只需保证 acks=all + enable.idempotence=true，不要试图靠分区语义做去重。

先质疑你的假设

1. Kafka 的 partitioner 是 murmur2(key) % numPartitions，对同一个 key 永远哈希到同一个分区，除非：
   • 你中途改过 partitioner / 换过序列化让 key 变了（比如 key 从 userId 变成 tenantId:userId）；
   • 或你用了 StickyPartitioner（无 key 时才生效，有 key 时仍走哈希）；
   • 或 broker 端 topic 配置了 partitioner.class 自定义实现且实现不幂等。 这三种情况之外，"同 key 因 broker 故障切换落到不同分区"在标准 Kafka 里不会发生。你压测里看到的"两个分区各写一次"更可能是消费组 rebalance 期间的重复消费：rebalance 时分区 A 被实例 1 持有、offset 尚未提交，实例 2 接管后从旧 offset 重读，同一批消息被两个实例各写一次——这是 at-least-once 的预期行为，不是 bug。
2. 0.3% 的"旧 version 覆盖新值"更像是并发写竞争：两个消费线程同时处理同一 userId 的 v5 和 v6，v5 的 UPDATE 在 v6 之后到达并提交，你的 WHERE version > 当前version 守卫在 v5 执行时看到当前 version=4，于是通过，把 v6 覆盖回 v5。这是典型的读-改-写竞态，不是消息乱序。

推荐组合（8000 msg/s 规模，可直接落地）

1. Kafka producer 侧（只做可靠性，不做去重）：

   acks=all
   enable.idempotence=true          # 0.11+，KIP-98
   retries=Integer.MAX_VALUE
   max.in.flight.requests.per.connection=5   # 幂等开启后允许 >1
   linger.ms=5
   batch.size=131072
   

   幂等 producer 保证单分区内不重发、不乱序，把 broker 侧的重复降到 0。

2. Kafka consumer 侧（保证 offset 与业务写入原子）：

   // 手动提交 offset，业务写入成功后才 commit
   consumer.commitAsync(offsets, (offsets, ex) -> {
       if (ex != null) log.error("commit failed", ex);
   });
   

   关键：不要 auto.commit，且 commit 必须在 UPDATE 成功之后。这样崩溃时最多重放"已写入但未 commit"的一小批，由下游幂等兜住。

3. MySQL 侧（真正的幂等根基）：

   -- 表结构：把 version 提为唯一键的一部分
   ALTER TABLE user_profile
     ADD UNIQUE KEY uk_user_version (user_id, version);
   
   -- 写入：用 INSERT ... ON DUPLICATE KEY 或 CAS
   INSERT INTO user_profile (user_id, version, payload, updated_at)
   VALUES (?, ?, ?, NOW())
   ON DUPLICATE KEY UPDATE
     payload    = IF(VALUES(version) > version, VALUES(payload), payload),
     updated_at = IF(VALUES(version) > version, NOW(), updated_at);
   

   这样即使同一 (userId, version) 被写两次，第二次是 no-op；即使 v5 在 v6 之后到达，IF 条件拦住，不会覆盖。

   如果你不想改表结构，退一步用行级 CAS + 行锁：

   UPDATE user_profile
   SET payload = ?, version = ?
   WHERE user_id = ? AND version < ?;
   -- affected_rows == 0 即表示已被更新版本覆盖，丢弃即可
   

   配合 innodb_lock_wait_timeout=5 和消费端按 userId 分桶串行化（见下）。

4. 消费端串行化（根治并发写竞态的关键）：

   • 用 userId.hashCode() % 12 把消息路由到 12 个固定消费线程之一（与消费组实例数对齐）；
   • 每个线程内单线程顺序处理该桶的消息，天然消除同 userId 的并发写；
   • 不同 userId 之间仍并行，吞吐不降。

   // 伪代码：按 userId 分桶
   int bucket = Math.abs(userId.hashCode()) % 12;
   bucketExecutors[bucket].submit(() -> handle(msg));
   

   这一步做完，你 0.3% 的"旧覆盖新"应该直接归零——因为同 userId 不再有并发写。

取舍与失败模式 | 方案 | 优点 | 失败模式 | |---|---|---| | (userId, version) 唯一键 | 幂等最强，重复写天然 no-op | 表结构变更需 DDL；version 必须严格单调 | | CAS + 行锁 | 不改表 | 高并发下锁等待；需配合桶化串行化 | | 桶化串行化 | 消除并发写竞态，吞吐不降 | 单桶热点（某个 userId 流量极高）会成瓶颈 | | Kafka 事务 | 跨 topic 原子 | 本场景单 topic 用不上，复杂度不值 |

边界条件：以上组合在 at-least-once 下是最终一致的——崩溃后最多重放一小批未 commit 的消息，由下游幂等兜住；不会丢消息，不会覆盖新值。如果业务要求严格一次（exactly-once），Kafka 原生只支持 sink 是 Kafka 的场景（KIP-98 事务），对 MySQL sink 不存在原生 exactly-once，只能靠"at-least-once + 幂等"逼近，这也是业界标准做法。

量化预期：8000 msg/s、32 分区、12 消费实例，桶化后每桶 ~667 msg/s，单线程处理无压力；MySQL 侧 (userId, version) 唯一键的 upsert 在 8000 QPS 下对 16C64G 的 RDS 是轻负载（预计 CPU < 15%）。改造后"旧覆盖新"应从 0.3% 降到 0，重复消息仍会有（at-least-once 预期），但全部被幂等键吃掉，下游看到的效果等价于 exactly-once。

一句话总结：Kafka 只管"不丢、不乱序（单分区内）"，幂等是下游的事——把 (userId, version) 做成唯一键 + 消费端按 userId 桶化串行化，0.3% 的覆盖问题就根治了，不需要碰分区语义。

先给结论：你的"跨分区重投"假设不成立，Kafka producer 的 partitioner 对同一 key 是确定性的（默认 murmur2(key.hashCode()) % numPartitions），broker 故障切换不会改变 key→partition 映射，同一 key 的消息永远落同一分区。你真正的问题是消费端时序倒置——12 实例消费组在 rebalance 窗口 + 处理耗时差异下，同一 userId 的旧 version 消息在新 version 消息落库之后才执行 UPDATE，version 守卫的 WHERE version > 当前 在并发下拦不住"旧值后写"。

一、先把根因说透

Kafka 保证的是分区内 offset 有序，不保证跨实例处理有序。你的 0.3% 回滚来自以下任一路径（大概率是 1+3 叠加）：

1. Rebalance 窗口时序倒置：实例 A 持有分区 P，正在处理 offset 100~200 的消息（含 userId=X, version=10），此时 rebalance 触发，分区 P 转给实例 B。实例 B 从 committed offset（比如 150）开始消费，先处理了 version=11 并落库。但实例 A 还没来得及 commit 的 version=10 消息，在 A 的线程里继续执行了 UPDATE——此时库里已经是 version=11，WHERE version > 10 不满足，看似被挡住了；但如果 A 的 UPDATE 先于 B 的 UPDATE 执行（网络/锁竞争），version=10 先落库，version=11 后落库，最终正确。问题是反过来：如果 A 的 version=10 因为 MySQL 行锁等待，在 B 的 version=11 落库之后才拿到锁执行，WHERE version > 10 此时 10 < 11，被挡住，没问题。那 0.3% 怎么来的？

   答案是：你的 version 守卫 SQL 大概率是 UPDATE ... SET version=10 WHERE user_id=X AND version < 10（注意是 < 不是 >），或者更常见的是 INSERT ... ON DUPLICATE KEY UPDATE version=VALUES(version) 没带 version 条件——后者会让旧值无条件覆盖。请立刻检查你的 upsert SQL。

2. Consumer 端没有保证同一 key 串行：如果你用的是 assign() 手动分配 + 多线程消费（一个分区分配给多个线程），同一分区的消息会被并发处理，version 守卫在并发下失效。

3. Producer 端重复发送：acks=all + retries 下，如果 producer 在 send() 的 ack 响应丢失后重试，同一消息会发两次（at-least-once 的语义），落到同一分区的相邻 offset，消费端处理两次。version 相同，幂等没问题；但如果 producer 端有"补偿重发"逻辑（比如本地消息表补偿），补偿的消息 version 可能比正常链路的小（因为补偿是旧消息），这就是"旧值后到"的真正来源。

二、根治方案（按优先级）

方案 1：Consumer 端 (user_id, version) 唯一键 + 严格 CAS（推荐，改动最小）

核心思路：不靠"WHERE version > 当前"这种比较守卫，而是靠数据库唯一约束做硬屏障。

-- 表结构改造：加 (user_id, version) 唯一索引
ALTER TABLE user_profile
  ADD UNIQUE INDEX uk_user_version (user_id, version);

-- Upsert 改为：先 INSERT，冲突时忽略（旧值无法覆盖新值）
INSERT INTO user_profile (user_id, version, payload, updated_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
  payload = IF(VALUES(version) > version, VALUES(payload), payload),
  updated_at = IF(VALUES(version) > version, NOW(), updated_at);

为什么这能根治：(user_id, version) 唯一键保证同一 user 的同一 version 只能插入一次（重复消息被唯一键挡住）；ON DUPLICATE KEY UPDATE 里的 IF(VALUES(version) > version, ...) 保证只有更新的 version 才能覆盖旧值，旧值后到会被静默忽略。两个条件叠加，无论消息以什么顺序到达、被哪个实例处理，最终库里一定是最大 version 的值。

失败模式：

• 如果业务上同一个 user 的 version 不是严格单调递增（比如两个上游各写一个 version 序列），这个方案失效，需要改成 (source_id, version) 唯一键 + 按 source 分别 CAS。
• MySQL 8.0 的 ON DUPLICATE KEY UPDATE 在高并发下会有行锁竞争，8000 msg/s 峰值下单表 upsert 的 TPS 上限约 3000~5000（取决于行大小和索引数量），你的 8000 msg/s 需要分表（按 user_id 哈希分 4~8 张表），或者换 TiDB/CockroachDB 这类原生分布式数据库。

代码示例（Java，Kafka Consumer）：

@KafkaListener(topics = "user-profile-event", groupId = "profile-aggregate")
public void onMessage(ConsumerRecord<String, ProfileEvent> record) {
    ProfileEvent event = record.value();
    jdbc.update(
        "INSERT INTO user_profile (user_id, version, payload, updated_at) " +
        "VALUES (?, ?, ?, NOW()) " +
        "ON DUPLICATE KEY UPDATE " +
        "payload = IF(VALUES(version) > version, VALUES(payload), payload), " +
        "updated_at = IF(VALUES(version) > version, NOW(), updated_at)",
        event.getUserId(), event.getVersion(), event.getPayloadJson()
    );
}

关键：@KafkaListener 默认是分区内单线程消费（一个分区同一时刻只有一个线程处理），这是 version 守卫生效的前提。如果你用了 concurrency 参数或手动 assign() + 线程池，必须保证同一分区的消息串行处理，否则唯一键 + CAS 也救不了并发下的时序倒置。

方案 2：Producer 端幂等 Producer（KIP-98）+ 事务（更强一致性）

如果你用的是 Kafka 3.x，开启幂等 producer：

# producer.properties
enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5  # 幂等模式下可 >1
retries=2147483647

幂等 producer 的效果：同一 PID（Producer ID）+ 同一序列号的消息，broker 端去重，网络层重复发送不会导致 broker 端重复存储。这解决的是"producer 端 at-least-once 导致的重复投递"，但不解决消费端时序倒置——如果消费端本身有并发问题，幂等 producer 帮不上忙。

Kafka 事务（transactional.id + initTransactions()）可以把"写下游 MySQL + 提交 Kafka offset"包进一个原子操作，但要求下游也接入 Kafka 事务（通过 Outbox 模式），改造成本大，8000 msg/s 规模下不建议，除非你有强一致诉求（如资金链路）。

方案 3：按 key 路由 + 单分区单线程消费（最稳，但有吞吐上限）

// 手动分配，保证同一 user 的消息始终由同一实例处理
List<TopicPartition> partitions = getPartitionsForUser(userId);
consumer.assign(partitions);

配合方案 1 的 SQL，同一 user 的消息在消费端绝对串行，version 守卫 100% 生效。

取舍：

• 优点：彻底消除并发时序倒置，version 守卫可靠。
• 缺点：单实例消费能力受限于单线程吞吐，8000 msg/s 需要 12 实例 × 每实例 667 msg/s，单线程 667 msg/s 的 MySQL upsert 是可以做到的（单条 upsert ~1ms），但要求你的消费逻辑（解析、校验、写库）总耗时 < 1.5ms/条。如果消费逻辑有外部 RPC 调用，单线程吞吐会掉到 100~200 msg/s，12 实例只能扛 1200~2400 msg/s，峰值 8000 会积压。

三、8000 msg/s 规模下的推荐组合

方案 1（唯一键 + CAS）+ 分表（4~8 张）+ 分区内单线程消费，理由：

1. 唯一键 + CAS 是数据库层硬屏障，不依赖应用层逻辑，最不容易出 bug；
2. 分区内单线程是 Kafka 消费默认行为，不需要改代码（只要不用 concurrency 参数或手动线程池）；
3. 分表解决单表 upsert TPS 瓶颈，8000 msg/s / 8 表 = 1000 msg/s/表，MySQL 单表 1000 upsert/s 轻松；
4. 不需要 Kafka 事务，因为你的诉求是"最终一致"而非"强一致"，version 守卫已经能保证最终落到最大 version 的值。

边界条件：

• 如果业务上允许"短暂读到旧值"（比如 1~2 秒），方案 1 够用；
• 如果业务上要求"任何时刻读到的值都是最新"（比如实时风控），方案 1 不够，需要 Kafka 事务 + Outbox，或者换 Redis + 版本号做前置去重。

四、立刻可以做的 3 件事

1. 检查你的 upsert SQL：如果是 ON DUPLICATE KEY UPDATE version=VALUES(version) 没带条件，改成方案 1 的 IF(VALUES(version) > version, ...) 写法，改完跑压测，0.3% 回滚大概率直接消失；
2. 确认消费端是分区内单线程：检查 @KafkaListener 的 concurrency 参数（默认 1，没问题）或手动 assign() 后是否起了线程池（如果有，去掉）；
3. 加监控：在消费端打一个指标 user_profile_upsert_ignored_total（IF 条件不满足时计数），压测时看这个指标的增量，它就是你被唯一键 + CAS 挡住的旧值数量，如果增量为 0 说明没有旧值后到，如果增量 > 0 说明还有并发问题需要排查。

回到你的"跨分区重投"假设：不成立。Kafka 的 partitioner 是确定性的，同一 key 永远落同一分区。你的 0.3% 回滚大概率是 upsert SQL 没带 version 条件 + 消费端并发 的叠加，不是 broker 端的问题。先把 SQL 改了，再跑一轮压测，大概率能复现到 0。