Kafka 事务与 Exactly-Once 的真实代价

"Exactly-Once"是消息队列领域最容易被过度承诺的词。Kafka 确实提供事务与幂等生产者，但它给的是在特定边界内的恰好一次语义，越界使用就会退化成 at-least-once，甚至产生数据错乱。本文说清这套机制的实际保证范围与代价。

一、三个层次别混为一谈

幂等生产者（enable.idempotence=true）：解决单分区、单会话内的重复写入（重试导致的重复）。它不跨越生产者重启，也不跨分区。

事务：让"多分区写入 + 消费位点提交"原子化，实现 Read-Process-Write 的原子闭环。

端到端 exactly-once：只有当生产、处理、消费位点提交全部落在同一个 Kafka 事务边界内，并且下游也是 Kafka时才成立。一旦下游是数据库、HTTP 或外部系统，就退回到 at-least-once 加业务幂等。

这是最关键的判据：下游不是 Kafka，就不要指望 Kafka 事务帮你实现端到端恰好一次。

二、幂等生产者

enable.idempotence=true          # 新版本默认开启
acks=all
retries=int32 最大值
max.in.flight.requests.per.connection=5
delivery.timeout.ms=120000

实现思路是给每个分区分配 producer ID 与单调递增的 sequence number，broker 侧去重旧序号。代价是吞吐下降（通常 5% 上下）与延迟略升，性价比很高，生产环境应当默认开启，它挡掉的是最常见的"重试导致重复"。

限制要清楚：PID 与序号缓存在生产者实例内存里，进程重启后序号重新开始，因此无法防止重启前的消息在重启后重复。跨会话去重要靠事务或业务层。

三、事务的原子边界

典型形态是"消费一批、处理后写回，并原子提交 offset"：

producer.initTransactions();
while (true) {
    ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofMillis(100));
    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> r : records) {
            producer.send(new ProducerRecord<>("out", transform(r)));
        }
        producer.sendOffsetsToTransaction(
                currentOffsets(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        producer.close();                       // 不可恢复，必须重建
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();
    }
}

消费者必须 isolation.level=read_committed（写回同一 topic 时尤其重要，否则会读到未提交数据）。

事务的原子性来自事务协调器（Transaction Coordinator）写入事务标记（control batch）：commit 后下游才可见，abort 后消息会被控制记录标记为无效。但 abort 不回收磁盘空间，这些记录仍占存储直到日志保留期结束。

四、代价清单

事务不是免费语义，四类代价必须评估：

1. 延迟增加。每次 commit 都要与协调器往返并写事务标记到分区，端到端延迟增加数十毫秒量级；批量越小，相对开销越大；
2. 吞吐下降。事务提交期间分区写入被串行化，transaction.timeout.ms 默认 1 分钟，长事务会阻塞该生产者；
3. transactional.id 的串行语义。相同 transactional.id 的新生产者启动时会 fence 掉旧的（触发 epoch 递增与旧事务回滚）。这意味着同一 transactional.id 不能并发多实例——多实例要按分区/桶唯一化 transactional.id，这正是 Flink 等框架为每个 subtask 分配不同 id 的原因；
4. 消费者侧的额外复杂度。read_committed 需要保留未决定事务的窗口，max.poll.interval.ms 与 transaction.timeout.ms 要协调设置，否则长事务会让消费者被判超时踢出组。

五、transactional.id 分配规则

这是最容易配错的地方。三条规则：

• 每个逻辑写入单元（如 Flink subtask、每个 partition writer）一个独立的 transactional.id；
• 实例重启时用相同的 transactional.id，才能触发 fence 机制把上一次未结束的事务回滚，避免僵尸事务把下游卡在 read_committed 等待；
• 不要让多个进程实例共用同一个 id，否则它们会互相 fence，表现为随机失败与莫名回滚。

flink-order-job-0-task-7-partition-3

粗粒度地给整个应用配一个 transactional.id，等价于把并行度压成 1。

六、更常用的落地组合

真实项目里更常见的正确方案，其实是at-least-once + 幂等消费，而不是全链路事务：

-- 消费端幂等：唯一业务键 + 去重表 / UPSERT
INSERT INTO article_stats(article_id, view_count)
VALUES (?, ?)
ON DUPLICATE KEY UPDATE view_count = VALUES(view_count);

// Redis 去重：按消息唯一键设置短期标记
ok, _ := rdb.SetNX(ctx, "consumed:"+msg.ID, 1, 24*time.Hour).Result()
if !ok {
    return nil                       // 已处理过，直接确认
}

两种组合的取舍很直白：业务操作本身可幂等（UPSERT、去重表、状态机版本判断）就用 at-least-once，成本低、延迟低；只在"跨多个 Kafka topic 的原子写"或"必须精确一次输出到 Kafka"时才上事务。

还有一类正确方案是outbox 模式：业务写库与"待发消息"在同一个本地事务里落库，再由 relay 投递。它把"数据库与消息之间的原子性"问题转化成"数据库本地事务"问题，绕开了跨系统事务。

BEGIN;
  INSERT INTO articles(...) VALUES (...);
  INSERT INTO outbox(topic, payload, created_at) VALUES ('article.created', ?, now());
COMMIT;
-- relay 轮询 outbox 投递，投递成功后标记；下游靠幂等去重

小结

Kafka 事务的真实边界是"多分区写入与 offset 提交原子化"，它给不了跨异构系统的端到端 exactly-once。代价有四块：延迟与吞吐、transactional.id 的串行化与 fence 语义、abort 记录仍占空间、以及 read_committed 与消费超时的协调。落地判据：业务侧能幂等就用 at-least-once 加幂等消费（这是绝大多数场景的正解），需要跨系统原子就考虑 outbox；只有纯 Kafka 到 Kafka 的严格语义场景才值得付出事务的成本。