Exactly-Once 的挑战
流处理系统通常涉及三个操作：
- 从输入 Topic 消费消息
- 处理消息（更新状态）
- 向输出 Topic 生产结果
如果操作 3 成功后应用崩溃，重启后会重新消费操作 1 的消息，导致重复输出。
Kafka Streams 的 Exactly-Once（EOS）保证的是**「消费 Kafka → 更新状态 → 生产到 Kafka」这一段的处理语义**：故障恢复后，输出 Topic 中不会出现重复记录，状态也不会被重复更新。它的前提是输入输出都是 Kafka、下游用 read_committed 消费；往外部数据库或 Redis 写不在这个保证范围内。
EOS 配置
Properties props = new Properties();
// 只需要开这个开关
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
开启 EOS 后，Kafka Streams 会自己完成下面这些配置，不要手写：
| 配置 | Streams 自动设置的值 |
|---|---|
enable.idempotence | true |
transactional.id | EOS v2 下是 application.id-processId-threadId；旧的 V1 是 application.id-taskId |
isolation.level | read_committed |
transaction.timeout.ms | 10 秒（commit.interval.ms 必须小于它） |
在 PRODUCER_PREFIX 下自己塞 transactional.id 会被 Streams 覆盖，容易让人误以为它可控；多个实例也不会靠你拼的字符串区分，EOS v2 由 Streams 按 processId 与线程自动分配（V1 才是按 task）。
EOS 工作原理
Kafka EOS 基于事务和幂等生产者：
消费批次 [offset 100-109]
↓
处理（更新状态存储）
↓
生产输出到 Topic（未提交）
↓
提交事务：
- 发送消费 offset 到 __consumer_offsets
- 发送输出消息（变为可读）
- 原子完成
关键：消费 offset 和生产输出在同一个 Kafka 事务中提交。要么都成功，要么都回滚。
两阶段提交
Phase 1（准备）：
Producer.beginTransaction()
处理消息，produce 到输出 Topic
Producer.sendOffsetsToTransaction(消费位置)
Phase 2（提交）：
Producer.commitTransaction()
或 Producer.abortTransaction()（失败时）
消费者配置 isolation.level=read_committed，只能看到已提交的事务消息。未提交的事务消息对消费者不可见。
EOS 的局限性
| 限制 | 说明 |
|---|---|
| 吞吐下降 | 事务与提交有额外开销，下降比例取决于批次大小和 commit 间隔，要自己压测，不要引用固定百分比 |
| 延迟增加 | 结果在事务提交前对下游不可见 |
| 多输出 Topic | 一个事务可以写多个 Topic，这本身不是限制；下游必须以 read_committed 消费 |
| 事务超时 | Streams 默认 10 秒（通用 Producer 默认才是 60 秒），单批处理或 commit 间隔超过它就会失败 |
| rebalance 影响 | Consumer Group rebalance 会中止在途事务，恢复后从头处理该批 |
性能优化
// 想拉长 commit 间隔，必须同步放大事务超时；否则 Streams 启动时报
// commit.interval.ms 必须小于 transaction.timeout.ms
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 30_000);
props.put(StreamsConfig.producerPrefix(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG), 45_000);
// 增加批次大小（生产者参数，与事务无关）
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
// 使用 EOS V2（每个 StreamThread 一个事务并携带 Consumer Group 元数据，
// rebalance 时不必为整个应用重新 fence；V1 按 task 分配 transactional.id，已不推荐）
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
commit.interval.ms 越大，事务批次越大、吞吐越好、结果可见延迟越高。EOS 下默认值是 100ms，AT_LEAST_ONCE 下才是 30s，直接套用后者的 30s 很容易撞上 10s 的事务超时。
测试 EOS
@Test
void testTopologyOutput() {
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// ... 其他配置
try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
input.pipeInput("key1", "value1");
assertEquals("VALUE1", output.readValue());
}
}
TopologyTestDriver 在单线程内存里跑拓扑，没有 broker、没有事务协调器，它不会验证 EOS 语义——上面的测试只证明「拓扑算得对」。要验证「故障时不重复、不丢失」，必须用真实 Kafka：KafkaStreams 连接 Testcontainers 启动的 broker，在写入中途杀掉实例再重启，然后统计输出 Topic 的消息数。
小结
| 配置 | 作用 |
|---|---|
EXACTLY_ONCE_V2 | 开启 EOS V2 |
enable.idempotence | 生产者幂等，Streams 自动开启 |
transactional.id | EOS V2 下按 application.id-processId-threadId 自动生成（V1 为 application.id-taskId），不要手写 |
isolation.level=read_committed | Streams 自动设置；下游自己的消费者也要设 |
commit.interval.ms | 事务提交间隔，EOS 下默认 100ms，必须小于 transaction.timeout.ms |
EOS 不是银弹。只在需要强一致性保证时使用（如金融交易、对账系统）。日志分析、监控指标等场景用 AT_LEAST_ONCE 即可。
下一节 3.3 Kafka Streams 与 Reactive Streams 衔接 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。