《Java 消息中间件·可靠消息与消费治理》2.3 恰好一次的两种实现与成本

两种「恰好一次」说的不是一件事

「Kafka 支持恰好一次」这句话经常被当成「业务不会重复」。这是两个不同层次的承诺：

• Kafka 事务保证：一组生产记录和一组消费位移，要么全部可见，要么全部不可见。它解决的是 Kafka 内部的原子性。
• 业务恰好一次保证：外部副作用（写库、扣库存、发通知）只发生一次。它解决的是跨系统的一致性问题。

前者的边界在 Kafka 的日志和事务协调器里，后者的边界在业务存储和幂等逻辑里。事务可以帮你把「读 A 主题、写 B 主题、提交 A 的位移」变成一个原子动作，但它管不到你在处理过程中调用的一次 HTTP 或者一条 INSERT。

这一节先把事务覆盖到哪里讲清楚，再用两组实验说明成本和边界，最后给出选型表。

事务到底覆盖了什么

一个事务生产者从构造到提交，关键调用只有五个：

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:39192");
p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "libraryd-tx-cost");
// 4.3.0 的常量名是 TRANSACTION_TIMEOUT_CONFIG，不是 TRANSACTION_TIMEOUT_MS_CONFIG
p.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "20000");

KafkaProducer<String, String> producer = new KafkaProducer<>(p);
producer.initTransactions();
producer.beginTransaction();
try {
    producer.send(new ProducerRecord<>("library.loan.created", "book-1", event));
    producer.send(new ProducerRecord<>("library.inventory.adjusted", "book-1", adjust));
    producer.commitTransaction();
} catch (RuntimeException e) {
    producer.abortTransaction();
    throw e;
}

initTransactions() 向协调器注册 transactional.id，拿到 PID 和 epoch；beginTransaction() 在客户端本地开启事务；send() 的记录带上事务标记；commitTransaction() 发 EndTxn，协调器把事务标记为已提交。消费者设置 isolation.level=read_committed 后，只会读到已提交事务的记录，abort 的记录不可见。

覆盖范围有三个明确的边界：

1. 只覆盖这个生产者发送的记录。另一个生产者写同一主题的记录不在事务里，除非它也用同一个 transactional.id（通常不应该）。
2. 不自动包含消费位移。要让「读 A、写 B、提交 A 位移」原子化，必须用 sendOffsetsToTransaction() 把位移交给事务协调器；普通 commitSync() 是另一个动作。
3. 不包含任何外部系统。数据库、缓存、HTTP 调用都不在事务协调器的管辖范围内。事务提交成功只说明 Kafka 侧成功了。

第三点是本节的核心。很多人把「事务提交成功」当成「业务成功」，然后在重放或崩溃恢复时发现副作用重复了。

一次事务在协议上做了什么

把上面的五个调用摊到协议层，大致是四步：

beginTransaction() 本身不发请求，它只是把客户端状态切到「事务中」。真正的网络开销从第一次 send() 开始：要先登记分区；commitTransaction() 再发一次 EndTxn 并等待协调器确认。所以「每条一事务」至少比普通发送多两次往返，这也解释了下一节测到的延迟差。

abort 也不是「把记录删掉」。记录已经写进分区，abort 只是写一个控制标记；read_committed 消费者读到标记后跳过这些记录，read_uncommitted 消费者仍然能看到它们。磁盘上的数据要等清理策略处理，log-end-offset 也已经把这些记录算进去了。

实验一：事务的成本

实验在同一台机器、同一个单节点 KRaft 上做三组发送，每组 200 条，全部 acks=all、enable.idempotence=true、linger.ms=0。三组分别是：

• 普通生产者，每条 send().get() 等待确认；
• 事务生产者，每条一个事务（begin / send / commit）；
• 事务生产者，每 10 条一个事务（20 个事务发 200 条）。

事务生产者的关键配置：

p.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "libraryd-tx-cost-" + runId);
p.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "20000");
p.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "30000");

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 单节点 KRaft / 2026-10-29）：

== cost == n=200 plainMs=886 txPerRecordMs=1987 txBatch10Ms=222
== cost == avgMs plain=4.43 txPerRecord=9.935 txBatch10=1.11 (单次运行，随调度波动，不作基准)

三组数字的关系比绝对值更重要：

• 每条一事务约为普通发送的 2.2 倍。多出来的部分是每个事务的 EndTxn 往返和协调器状态更新。事务不是免费的开关，它把一次网络往返变成了至少两次。
• 十条一批的事务反而比普通逐条发送更快。这不是说事务更快，而是两组的等待模式不同：普通组每条都等确认，事务组先流水线发 10 条再等一次提交。把批次拉平后比较才有意义，本节没有做这个对照，所以不能得出「事务比非事务快」的结论。
• 这些数字是单次运行，受调度和机器负载影响很大，只能用来判断数量级，不能当基准。

成本的可控手段主要是批的大小。每条一事务适合「严格一次一业务动作」的场景；如果业务允许把 10 条事件放在一个事务里，提交开销被摊薄，延迟会明显下降。代价是失败时 10 条一起回滚，重试的粒度也变成 10 条。

成本不只在延迟

事务还有三项不容易在单次压测里看到的成本：

第二项最容易被忽略。read_committed 的可见性上界不是 log-end-offset，而是 last-stable-offset（LSO）。只要有一个事务长时间不提交，LSO 就停在那条记录之前，后来的已提交记录也读不到。表现为「生产明明成功了，消费者却停在原地」，而 lag 指标可能还在涨。排查时要同时看 LSO 和 open transaction 的数量，不能只看 log-end。

因此事务的第一个运维约束是事务要短。transaction.timeout.ms 不是「可以开多久」的许可，而是「超时会被协调器 abort」的兜底。业务处理超过这个时间，事务会被强制回滚，客户端后续调用收到超时或 fencing 错误。

实验二：事务提交了，业务还是重复了

这一组实验要证明的是：事务的「恰好一次」只到 Kafka 边界为止。

步骤：

1. 事务生产者在一个事务里写 5 条 library.loan.tx.business 记录，提交；
2. read_committed 消费者从头读，读到 5 条，把每条应用到两个投影：
   • naive：每来一条就加一次（模拟 stock = stock - 1 或追加一行）；
   • dedupeBySink：按 eventId 覆盖（模拟幂等写入）；
3. 模拟投影重建：seekToBeginning 再读一遍，同样应用一次；
4. 对比两个投影的最终状态。

消费者配置的关键一行：

props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

应用逻辑：

for (ConsumerRecord<String, String> r : records) {
    String eventId = field(r.value(), "eventId");
    naive++;                                   // 非幂等：重放就再加一次
    dedupeBySink.put(eventId, r.value());      // 幂等：同一个 eventId 覆盖
}

本机实测输出：

== business == committedRecords=5 firstPass=5 replayPass=5 naiveEffect=10 dedupeBySinkSize=5

事务提交了 5 条，read_committed 第一遍读到 5 条，重放又读到 5 条。Kafka 侧没有任何重复：log-end-offset 是 5，没有第二条相同 offset 的记录。但非幂等投影的效果变成 10，幂等投影仍然是 5。

这条链路说明三件事：

• 事务保证的是「这 5 条要么都可见、要么都不可见」，它不保证「这 5 条只被业务消费一次」；
• 重放是合法操作。新投影上线、补历史、换存储都会重读已提交记录，事务历史不会阻止它；
• 业务是否重复，取决于消费之后那一步怎么写，而不是生产端用没用事务。

如果这个消费者还要提交位移，并且希望在「处理 + 提交」之间也原子，可以用 sendOffsetsToTransaction() 把位移纳入事务。但即使这样，外部投影的写入仍然不在事务里：崩溃可能发生在「投影写完、事务未提交」之间，重启后重读，投影再写一次。要消掉这一步，仍然需要业务幂等。

两种方案对照表

两者不是替代关系。最稳的组合是「事务 + 幂等」：事务保证 Kafka 侧不重不漏，幂等保证外部副作用只发生一次。代价是两套机制都要维护。

成立的前提清单

Kafka 事务成立的前提

• transactional.id 稳定且每个生产者实例唯一。重启后复用同一个 ID 会触发 fencing，旧实例被踢出，这是防止僵尸写入的机制。
• 所有要纳入事务的记录都通过同一个生产者发送。中途换生产者，新记录不在原事务里。
• 消费端使用 read_committed。read_uncommitted 会看到 abort 的记录，也会看到未提交的数据。
• 消费位移通过 sendOffsetsToTransaction() 提交。用普通 commitSync() 提交，位移和记录就不在同一个原子单元里。
• transaction.timeout.ms 大于业务处理时间。超时后协调器会 abort，客户端再提交会失败。
• 生产环境的事务状态主题副本数至少 3。单节点 RF=1 只能做实验，broker 挂了事务状态就没了。

业务幂等成立的前提

• 事件 ID 在生产端生成，重试和重放时不变。
• 去重记录和业务写入在同一个事务里，或者去重存储本身能保证「占位 + 执行」的原子性。
• TTL 覆盖最大重投递窗口，包括 broker 保留期、消费者停机时间和下游重放窗口。
• 去重键能区分合法的重复业务（同一个 bookId 的两次借阅不是重复）。
• 并发重复投递被正确处理，不能是「先查再写」的竞态。
• 去重表满了有明确策略：拒绝、降级还是扩容，不能静默淘汰未过期项。

两份清单里，事务的前三条和幂等的前三条是最容易漏的。只要漏一条，「恰好一次」就退化成「至少一次」，而且往往在故障恢复时才暴露。

为什么事务也不等于业务恰好一次

把原因收拢成四条：

1. 外部副作用不在事务里。数据库写入、HTTP 调用、缓存更新都不受事务协调器管理。事务提交成功，外部调用可能失败；外部调用成功，事务可能回滚。
2. 消费位移可以不在事务里。即使记录是事务提交的，消费者处理完、提交位移前崩溃，重启后仍会重读。只有把位移也交给 sendOffsetsToTransaction()，这一段才是原子的。
3. 重放是设计内的操作。重建投影、补数、换存储都会重读已提交记录。事务保证这些记录只存在一份，不保证只被应用一次。
4. 跨主题、跨集群的边界。事务只覆盖注册过的分区。写到另一个集群、另一个主题（用不同生产者）的数据不在同一个事务里；跨集群复制也不保持事务原子性。

这四条里，第 1 条和第 3 条无法靠 Kafka 解决，只能靠业务幂等或对账。第 2 条可以用 sendOffsetsToTransaction() 缓解，但外部副作用仍然要单独处理。第 4 条是架构边界，卷三讲跨集群时会再展开。

一个可执行的判断流程

选型时按顺序问四个问题：

1. 副作用是否完全发生在 Kafka 内部？是，事务可以覆盖；否，进入第 2 问。
2. 副作用能否用唯一键或状态迁移表达？能，业务幂等优先；不能，进入第 3 问。
3. 是否必须先写库、再发消息？是，outbox 加幂等；否，进入第 4 问。
4. 延迟预算是否允许每个事务至少两次额外往返？否，批量事务或只做幂等；是，事务加幂等。

这个顺序的道理是：能用便宜手段解决的，不要先上事务。事务的成本是每笔固定的往返和协调器负载，幂等的成本是一次存储读写；两者的取舍应该由副作用的位置决定，而不是由「Kafka 支持 EOS（exactly-once semantics）」这个事实决定。

怎么选

一个常见的误判是「用了事务就可以不做幂等」。只要副作用出了 Kafka，这句话就不成立。反过来，「做了幂等就不需要事务」也不对：在 Kafka 到 Kafka 的流处理里，事务能避免中间主题出现半成品，幂等表反而不好放。

回到 libraryd

卷二后面的实现会同时用两种机制：

• TransactionalProducer 负责 library.loan.created 到 library.inventory.adjusted 的读改写闭环，位移通过 sendOffsetsToTransaction() 提交；
• IdempotentConsumer 负责写数据库、发通知这类外部副作用，键按 2.2 的表选择；
• Outbox（7.1）负责「数据库提交 + 事件发布」的一致性，它解决的是事务覆盖不到的那一段。

三者的分工可以画成一条线：事务管 Kafka 内部，幂等管外部副作用，outbox 管本地事务与消息发布的衔接。任何一段缺失，端到端就只能做到至少一次。

下一节进入事务的具体用法：transactional.id 怎么选、read_committed 消费者看到什么、abort 后数据是什么形态。

阅读导航：上一节 2.2 幂等消费：唯一键、状态机与去重表 ｜ 下一节 3.1 事务生产者与 read_committed