作者：江若衡｜8 年技术团队负责人｜分布式一致性与消息中间件方向 栈：Kafka 3.5.1（KRaft，3 broker，8C32G）、Spring Kafka 3.1.5、Java 17（G1）、MySQL 8.0.36 场景：清算事件流
settlement-event，24 分区，消费组 6 实例，峰值 3.2 万 msg/s，单条 1.4KB
一、现象：幂等表 TTL 设了 7 天，但 7 天这个数字是怎么来的？
去年 Q3 做清算链路重构时，我在 code review 里看到幂等表 settlement_dedup 的 TTL 配置是 7 天（dedup_ttl_days=7）。我问了写这段代码的同事，他的回答是："参考了网上 Kafka 幂等的最佳实践，一般设 7 天。"
我追问："你的重复消费窗口实际有多长？"他答不上来。
于是我们花了一个下午做了一次量化：在 6 个消费实例上开了 7 天监控，统计每次 rebalance 事件前后 committed offset 与 processed offset 的差值（即"已处理但未提交"的消息数），乘以消息处理耗时，得到实际重复消费窗口。
结果：
| 指标 | 数值 |
|------|------|
| 单批 max.poll.records | 200 |
| 单条处理 P50 | 120ms |
| 单条处理 P99 | 800ms（含下游清算 RPC） |
| 单批处理 P99 | 200 × 800ms = 160s |
| rebalance（eager）耗时 P99 | 8.2s |
| 新实例接管 + 重新拉取 P99 | 2.1s |
| 实际重复消费窗口 W_repeat P99 | ≈ 170s ≈ 2.8 分钟 |
| 幂等表 TTL 设置 | 7 天 = 604800s |
W_repeat 是 2.8 分钟，TTL 设了 7 天——安全系数高达 3558×。 这不是"保守"，是"拍脑袋"。7 天的 TTL 意味着幂等表里积累了 7 × 24h × 3.2万msg/s × 86400s ≈ 193 亿条记录（按峰值估算），MySQL 的 settlement_dedup 表膨胀到 4.7TB，B+ 树 4 层，点查 P99 从 0.3ms 涨到 12ms。
更关键的是：7 天的 TTL 掩盖了一个真正的问题——它让人误以为"只要 TTL 够长，幂等就一定安全"，从而忽略了 TTL 太短时的真实风险场景（后面第四节详述）。
二、原理：offset 提交粒度如何决定重复消费窗口的上下界
Kafka 的 at-least-once 语义建立在一条核心假设上：offset 提交滞后于消息处理。这条假设决定了"重复消费窗口"的宽度，而窗口的宽度由三个因素共同决定：
2.1 下界：单批消息的处理时间
max.poll.records=N 条消息从 poll() 返回到 commitSync() 完成，这段时间内所有消息都是"已处理但未确认"状态。如果此时实例崩溃（OOM、kill、节点宕机），新实例接管后从上一次提交的 offset 开始重新消费，整批 N 条消息都会被重放。
下界公式：
W_lower = N × T_process_P99
我们的数字：200 × 800ms = 160s。
2.2 上界：rebalance + 重新分配 + 重新拉取
当消费组发生 rebalance（实例被踢、扩缩容、broker 故障），分区被重新分配。在 Eager 协议下（KIP-415 之前），全组分区的归属都会被重排，所有实例的 in-flight 消息都会被重新分配。新实例接管后，从 __consumer_offsets 读取上次提交的 offset，重新拉取。
上界公式：
W_upper = W_lower + T_rebalance + T_reassign + T_refetch
= N × T_process_P99 + T_rebalance_P99 + T_reassign_P99 + T_refetch_P99
我们的数字：160s + 8.2s + 2.1s + 1.5s ≈ 172s ≈ 2.9 分钟。
2.3 提交粒度对窗口宽度的影响
不同的 offset 提交策略，直接改变 W_lower：
| 提交策略 | W_lower | 说明 |
|----------|-----------|------|
| auto.commit.interval.ms=5000（默认） | 5s × 消息速率 | 每 5s 提交一次，窗口 = 5s 内处理的消息数 × 单条耗时 |
| 手动 commitSync（每批） | N × T_process_P99 | 整批处理完才提交，窗口 = 单批处理时间 |
| 逐条 commitSync | T_process_P99（单条） | 窗口最小，但每条一次网络往返，吞吐降 60%+ |
| commitAsync（每批） | 同 commitSync | 有丢失风险（提交请求在崩溃前未到达 broker），不推荐用于 at-least-once |
关键洞察：提交粒度不是"性能 vs 正确性"的权衡，而是重复消费窗口宽度的直接控制变量。选 auto-commit 5s 看似"更频繁提交、更安全"，实际上它把窗口从 160s 压缩到了 5s × 3.2万msg/s / 6实例 ≈ 26667 条——如果单条处理 800ms，这 26667 条的处理时间是 26667 × 800ms ≈ 21334s ≈ 5.9 小时。等等，这个计算有问题——auto-commit 是按时间提交的，不是按消息数。让我修正：
auto.commit.interval.ms=5000 的真实语义是：每 5s 把当前已处理的消息的 offset 提交一次。所以窗口 = 5s 内处理的消息数 × 单条耗时。如果 5s 内处理了 3.2万/6 × 5 ≈ 26667 条，窗口 = 26667 × 800ms ≈ 21334s。
这显然不对——auto-commit 的窗口不应该是 5.9 小时。问题出在我把"消息速率"和"单条耗时"混在一起了。
正确的理解是：auto-commit 每 5s 提交一次，提交的是当前已处理到的 offset。如果 5s 内处理了 26667 条消息，那么这 5s 的窗口内，这 26667 条消息都是"已处理但未确认"状态。如果此时崩溃，新实例从 5s 前的 offset 开始重放，重放 26667 条。
所以 auto-commit 5s 的窗口 = 5s（时间维度），重放消息数 = 5s 内的处理量。
而 commitSync 每批 200 条 的窗口 = 200 条的处理时间 = 160s，重放消息数 = 200 条。
对比：
auto-commit 5s：窗口 5s，重放 26667 条commitSync 200 条：窗口 160s，重放 200 条
反直觉：auto-commit 的窗口更短（5s < 160s），但重放的消息数更多（26667 > 200）。哪个更"安全"？取决于你的幂等表是按消息 ID 还是按时间做判定：
- 按消息 ID：
commitSync 200 条更优（只重放 200 条，幂等表命中 200 次） - 按时间：
auto-commit 5s更优（窗口 5s，幂等表 TTL 只需覆盖 5s）
结论：提交粒度的选择必须与幂等表的判定依据联动设计。
三、具体做法：三步量化 + schema 设计 + 提交粒度选择
3.1 第一步：用监控数据计算 W_repeat
在我们的消费端埋点（基于 Micrometer + Prometheus），每 10s 上报以下指标：
# 每批处理时间（从 poll 返回到 commitSync 完成）
kafka_consumer_batch_duration_seconds:
- buckets: [0.1, 0.5, 1, 5, 10, 30, 60, 120, 180, 240]
- quantiles: [0.5, 0.9, 0.99]
# rebalance 事件耗时
kafka_consumer_rebalance_duration_seconds:
- quantiles: [0.5, 0.9, 0.99]
# 新实例接管 + 重新拉取耗时
kafka_consumer_reassign_duration_seconds:
- quantiles: [0.5, 0.9, 0.99]
# 已处理但未提交的 offset 差值（lag 的"微观版"）
kafka_consumer_uncommitted_offset_diff:
- quantiles: [0.5, 0.9, 0.99]
7 天监控后，用以下公式计算 W_repeat：
W_repeat_P99 = P99(batch_duration) + P99(rebalance_duration) + P99(reassign_duration)
= 160s + 8.2s + 2.1s
= 170.3s ≈ 2.8 分钟
3.2 第二步：设计幂等表 schema
我们的幂等表 settlement_dedup 最终 schema：
CREATE TABLE settlement_dedup (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
biz_id VARCHAR(64) NOT NULL COMMENT '业务唯一键（如 settlement_id）',
event_type VARCHAR(32) NOT NULL COMMENT '事件类型（如 SETTLE_SUCCESS）',
partition_id INT NOT NULL COMMENT 'Kafka 分区号',
offset BIGINT NOT NULL COMMENT 'Kafka offset（用于按 offset 清理）',
processed_at DATETIME(3) NOT NULL COMMENT '处理时间（毫秒精度）',
result_code VARCHAR(16) NOT NULL COMMENT '处理结果码（SUCCESS/FAIL/SKIPPED）',
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
UNIQUE KEY uk_dedup (biz_id, event_type, partition_id, offset),
KEY idx_processed_at (processed_at),
KEY idx_offset (partition_id, offset)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='清算事件幂等表';
关键设计点：
- 唯一键包含
partition_id + offset：同一biz_id + event_type在不同 offset 出现时（rebalance 后重放），视为不同记录，不做幂等短路。这避免了"幂等表误判为已处理"的坑（见第四节坑 1）。 idx_offset索引：支持按 offset 清理（见 3.4）。result_code字段：区分"处理成功"和"处理失败"。失败的消息在重试时应该重新处理，而不是被幂等表短路。
3.3 第三步：设置 TTL
TTL 的计算公式：
TTL = W_repeat_P99 × safety_factor + manual_replay_window
其中：
W_repeat_P99 = 170ssafety_factor = 10（覆盖 P99.9 慢消息 + 网络抖动 + GC STW）manual_replay_window = 24h（覆盖人工重投窗口，通常运维在 24h 内完成）
TTL = 170s × 10 + 24h = 1700s + 86400s = 88100s ≈ 24.5h
最终 TTL 设为 25 小时（取整），而不是 7 天。
3.4 清理策略：按 offset 清理，而不是按时间清理
-- 每 10 分钟执行一次，清理已提交 offset 之前的幂等记录
DELETE FROM settlement_dedup
WHERE partition_id = ? AND offset < ?
LIMIT 10000;
为什么按 offset 清理而不是按 processed_at 清理？
- 按时间清理：
processed_at < NOW() - INTERVAL 25 HOUR。问题：如果某条消息在 24h 前处理成功，但 25h 后因为人工重投又被消费了一次，此时幂等表里的记录可能已经被清理了 → 幂等失效。 - 按 offset 清理：只要
committed offset还没超过这条记录的 offset，这条记录就不会被清理。即使 25h 后人工重投，只要 offset 还在committed offset之前，幂等表仍然有效。
但按 offset 清理也有前提：committed offset 必须是单调递增的（Kafka 保证），且清理线程要感知 committed offset 的最新值（通过 __consumer_offsets 或消费端上报）。
四、踩过的坑（三个反直觉点）
坑 1：幂等表只记 biz_id 不记 offset → rebalance 后重放被误判为"已处理"
场景：消息 settlement_id=ABC, event_type=SETTLE_SUCCESS 在 offset 1000 处理成功，幂等表记录了 (ABC, SETTLE_SUCCESS)。rebalance 后，同一条消息在 offset 1000 被重放（因为 offset 1000 还没提交）。消费端查幂等表，命中 (ABC, SETTLE_SUCCESS) → 直接 ACK，跳过处理。
问题：如果第一次处理时下游清算 API 调用成功但写库失败（事务回滚），那么"已处理"其实不是真的"已处理"。重放时应该重新处理，而不是被幂等表短路。
修复：幂等表唯一键加入 partition_id + offset，且 result_code 字段区分 SUCCESS/FAIL。只有 result_code=SUCCESS 的记录才做幂等短路；result_code=FAIL 的记录在重放时重新处理。
坑 2：TTL 太短 + 慢消息 → 幂等表过期后同一条消息再来，被误放行
场景：TTL 设为 10 分钟（覆盖 W_repeat=170s 的 3.5× 安全系数）。一条消息在 offset 2000 处理时，下游清算 API 超时（P99.9=8s），重试 3 次后成功，总耗时 30s。处理完成后 9 分钟，同一条消息因为 rebalance 被重放（offset 2000 还没提交）。此时幂等表里的记录已经过期（10 分钟 TTL），被清理 → 幂等失效，重复处理。
修复：TTL 必须覆盖 W_repeat + 最慢消息的处理时间 + 安全系数。我们的最慢消息 P99.9=8s，重试 3 次后总耗时 24s，所以 TTL 至少要覆盖 170s + 24s + 安全系数。最终我们设 TTL=25h（覆盖人工重投窗口），彻底避免这个问题。
坑 3：commitAsync 看起来"更快"，实际上有丢失风险
场景：为了提升吞吐，把 commitSync 改成 commitAsync。理论上 commitAsync 不阻塞消费线程，吞吐提升 15%。但实际上，commitAsync 的提交请求是异步的，如果实例在提交请求到达 broker 之前崩溃（OOM、kill），这次提交就丢失了 → 新实例从上一次成功提交的 offset 开始重放 → 重复消费窗口扩大。
更隐蔽的问题：commitAsync 的提交顺序不保证。如果连续两次 commitAsync(offset=1000) 和 commitAsync(offset=1005)，broker 可能先收到 1005 再收到 1000（网络乱序），最终 __consumer_offsets 里记录的是 1000 → offset 回退 → 重放 1000-1005 的消息。
修复：at-least-once 语义下，只用 commitSync。吞吐损失 15% 可以接受，但 offset 回退导致的重复消费是不可接受的。
五、结论与可复用清单
5.1 核心结论
- 幂等表 TTL 不是"拍脑袋"的参数，而是可以从监控数据算出来的。公式：
TTL = W_repeat_P99 × safety_factor + manual_replay_window。 - 提交粒度与幂等表判定依据必须联动设计。按消息 ID 判定 → 用
commitSync 每批；按时间判定 → 用auto-commit。 - 幂等表 schema 必须包含
partition_id + offset，否则 rebalance 后重放会被误判为"已处理"。 - at-least-once 语义下，只用
commitSync。commitAsync的"吞吐提升"是假象，实际引入了 offset 回退风险。
5.2 可复用清单
- [ ] 在消费端埋点
batch_duration、rebalance_duration、reassign_duration三个指标，跑 7 天监控 - [ ] 用
W_repeat_P99 = P99(batch) + P99(rebalance) + P99(reassign)计算重复消费窗口 - [ ] 幂等表唯一键包含
biz_id + event_type + partition_id + offset - [ ] 幂等表加
result_code字段，区分 SUCCESS/FAIL - [ ] TTL =
W_repeat_P99 × 10 + 24h（覆盖人工重投窗口） - [ ] 清理策略按 offset（不是按时间），每 10 分钟清理一次
- [ ] at-least-once 语义下只用
commitSync，不用commitAsync
5.3 开放问题
-
如果消息处理时间分布是重尾的（P99.9 比 P99 长 50 倍），
W_repeat的 P99 估计是否足够？ 我们的 P99.9=8s 是 P99=800ms 的 10 倍，安全系数 10× 刚好覆盖。但如果 P99.9 是 P99 的 100 倍（长尾更重），安全系数 10× 就不够了。有没有团队用分位数回归（而不是固定安全系数）来定 TTL？ -
auto-commit的"按时间提交"和commitSync的"按批提交"，在幂等表 schema 上有什么本质区别？ 我现在的理解是：auto-commit的幂等表只需要biz_id + event_type（因为窗口短，重放概率低），commitSync的幂等表需要biz_id + event_type + partition_id + offset（因为窗口长，rebalance 后重放概率高）。但这个区别有没有被其他团队验证过？ -
KIP-415（Cooperative Rebalancing）对 W_repeat 的影响是什么？ Eager 协议下 rebalance 耗时 8.2s，Cooperative 协议下是否更短？如果更短，W_repeat 的 P99 会降多少？有没有团队量化过？
-
幂等表的
result_code=FAIL记录在重放时"重新处理"，但如果第二次处理仍然 FAIL，第三次重放时怎么办？ 是无限重试还是进入 DLQ？DLQ 的消息再被人工重投时，幂等表里的result_code=FAIL记录是否应该被清理？
参考引用：
- 江清和《Kafka 消费者组 rebalancing 风暴：一次 200 万重复消息事故的根因与治理》（id=62）中关于
max.poll.interval.ms与 rebalance 风暴的量化分析 - 江景行《下游 RPC P99 抖到 1.8s 后消费组每 10 分钟被踢一次：Kafka 消费端背压的四层降级与量化参数》（id=1643）中关于
max.poll.interval.ms是"消费吞吐的硬预算"而非"超时保护"的论述