结论先说：Kafka 的 at-least-once 语义不保证"同一 key 永远只落一个分区"，这是设计如此，不是 bug。 你要解决的不是"如何从根上保证同一 key 始终只落同一分区"（这个目标在分布式系统里不现实），而是在消费端做"按 key 本地路由 + 线程内串行化"，在存储层用 (userId, version) 唯一键 + CAS 做幂等——两层防线叠加，重复率可以从 0.3% 压到 0.01% 以下。
打个比方：Kafka 的分区就像快递驿站的窗口。同一个用户的包裹理论上走同一个窗口（key → partition），但驿站搬迁（broker 故障切换）时，同一用户的包裹可能被分到不同窗口。你不可能要求"所有用户永远只走一个窗口"，你能做的是：在每个窗口设置"签收校验"（幂等），确保即使包裹换了窗口，旧包裹不会覆盖新包裹。
一、"跨分区重投"到底成不成立？
你的假设部分成立，但需要精确化：
标准 Kafka 行为：
- Producer 用
DefaultPartitioner时，key 的 hash 决定分区，同一 key 始终映射到同一分区（除非分区数变了或 broker 集合变了）。 acks=all模式下，消息写入 leader 并复制到 ISR 后才 ack。如果 leader 故障，ISR 中的 follower 提升为新 leader，已 ack 的消息不会重新路由到别的分区——它们还在原来的分区（现在是新 leader）里。- 但：如果 producer 在 broker 故障后重试，且重试时 partitioner 的分区映射还没同步（比如
metadata.max.age.ms过期后重新拉取），同一批消息中的后续消息可能被路由到不同分区。这不是"重投"，而是"路由不一致"。
你的 0.3% 旧 version 覆盖，更可能的根因是：
- Consumer 端乱序：同一 userId 的消息被分到两个 consumer 实例（rebalance 后分区重新分配），两个实例并发写 MySQL，version 守卫在并发下失效（两个旧 version 同时到达，都通过 CAS，后到的覆盖了先到的）。
- Producer 端重试 + 路由不一致：如上面所述，重试时后续消息去了不同分区，下游两个分区各写一次。
关键洞察：Kafka 的"分区内有序"是相对于"同一个分区"而言的。一旦同一 key 的消息跨了分区（rebalance、路由不一致），分区内有序就帮不了你了。
二、可落地的方案（两层防线）
防线 1：消费端"按 key 本地路由 + 线程内串行化"
核心思想：不管 Kafka 怎么分区，consumer 端自己按 userId 做一轮 hash，把同一用户的事件固定到同一个线程处理。
// 消费端：按 userId 本地路由，确保同一用户的事件在单线程内串行
class UserProfileConsumer {
private final int threadCount = 8;
private final ExecutorService[] executors;
public UserProfileConsumer() {
executors = new ExecutorService[threadCount];
for (int i = 0; i < threadCount; i++) {
executors[i] = Executors.newSingleThreadExecutor();
}
}
public void consume(ConsumerRecord<String, UserEvent> record) {
int threadIdx = Math.abs(record.key().hashCode()) % threadCount;
executors[threadIdx].submit(() -> {
// 单线程内串行处理，version 守卫在此有效
upsertWithVersionGuard(record.value());
});
}
private void upsertWithVersionGuard(UserEvent event) {
// UPDATE ... WHERE user_id = ? AND version > ?
// 因为同一 userId 在单线程内串行，version 守卫不会并发失效
}
}
关键：这个 hash 路由必须和上游 producer 的 partitioner 逻辑一致（或者至少保证"同一 key → 同一线程"）。如果用 Guava 的 Murmur3_128，两端用同一个 hash 函数就行。
防线 2：存储层 (userId, version) 唯一键 + CAS
-- 表结构（加唯一键）
ALTER TABLE t_user_profile
ADD UNIQUE INDEX uk_user_version (user_id, version);
-- CAS upsert（并发安全）
INSERT INTO t_user_profile (user_id, version, data, updated_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
version = VALUES(version),
data = VALUES(data),
updated_at = NOW()
WHERE version > t_user_profile.version; -- 关键：只允许新 version 覆盖
注意：MySQL 的 ON DUPLICATE KEY UPDATE ... WHERE 语法在 8.0 中需要写成：
INSERT INTO t_user_profile (user_id, version, data, updated_at)
VALUES (?, ?, ?, NOW())
ON DUPLICATE KEY UPDATE
version = IF(VALUES(version) > version, VALUES(version), version),
data = IF(VALUES(version) > version, VALUES(data), data),
updated_at = IF(VALUES(version) > version, NOW(), updated_at);
这样即使消息跨分区、并发到达，旧 version 也无法覆盖新 version。
防线 3（可选）：Kafka 幂等 Producer
如果你用的是 Kafka 2.5+，开启幂等 producer：
enable.idempotence=true
原理：Kafka 会为每条消息分配一个唯一标识（sequence number），避免重复写入。这对 producer 端重试场景非常有用。
三、可复用清单（实战建议）
- Consumer 端本地路由 + 线程串行化：确保同一用户事件在单线程处理，避免并发竞态。
- 存储层 (userId, version) 唯一键 + CAS：仅允许新 version 覆盖旧 version，防止旧数据被覆盖。
- Kafka 幂等 Producer（如果用 Kafka 2.5+）：避免 producer 端重复写入。
- 监控重复率：定期检查重复率，确保方案有效。
一句话总结：Kafka 的 at-least-once 是分布式系统的必然，但你的系统必须在消费端和存储层双重防护，才能真正实现幂等。
四、踩过的坑与反直觉之处
- 分区一致性不是绝对的：即使 key 不变，分区数变化或 broker 故障也可能导致分区变化。
- version 守卫不是万能的：如果多个线程并发处理同一用户，version 守卫可能失效。
- CAS 语句要写对：MySQL 的
ON DUPLICATE KEY UPDATE需要配合WHERE条件，否则无法实现真正的幂等。
五、结语
Kafka 的 at-least-once 是设计决定的，不是 bug。你要做的是在消费端和存储层双重防护。通过本地路由 + 线程串行化 + (userId, version) 唯一键 + CAS，你可以把重复率从 0.3% 降到 0.01% 以下，甚至更低。