要保证每条数据恰好处理一次——Flink的Exactly-Once怎么实现的？
总结
Flink的Exactly-Once分引擎内部（Barrier对齐）和端到端（两阶段提交）。Unaligned Checkpoint解决Barrier对齐的延迟问题。端到端需要Source支持重放 + Sink支持事务。面试时能区分这两层、说清楚两阶段提交的过程、再加一个Kafka事务超时的踩坑，就很有深度。
下一篇聊一个很实用的选型问题：Flink SQL和DataStream API该用哪个？什么场景必须用DataStream？
你们生产环境用的是At-Least-Once还是Exactly-Once？评论区聊聊～
关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长
先把这段讲顺；面试官追问时，再展开下面的原理和细节。
详细讲解
Exactly-Once是Flink的王牌能力，也是它比Spark Streaming强的核心原因之一。但很多人对Exactly-Once的理解停留在“开了Checkpoint就是Exactly-Once”，这是不对的。今天把Flink的Exactly-Once的实现原理和“端到端”的难点彻底讲清楚。
先说结论
Flink的Exactly-Once分两层：
引擎内部： 通过Checkpoint + Barrier对齐保证。这是Flink自己搞定的。
端到端（Source到Sink）： 需要Source支持重放（如Kafka）+ Sink支持事务或幂等写入。这需要整个链路配合。
引擎内部的Exactly-Once：Barrier对齐
第1篇已经讲了Checkpoint的过程。这里重点说“Barrier对齐”为什么能保证Exactly-Once：
当一个算子有多个输入流时，它要等所有输入流的Barrier都到齐才做Checkpoint。在等待期间，已到达Barrier的流的数据会被缓冲（不处理），确保状态快照的一致性。
但这有个代价：Barrier对齐会引入延迟。如果某个输入流的Barrier迟迟不来，其他流的数据就一直被缓冲，导致延迟。
Flink 1.11+的Unaligned Checkpoint
为了解决Barrier对齐的延迟问题，Flink 1.11引入了Unaligned Checkpoint：不等Barrier对齐，而是把缓冲中的数据也保存到Checkpoint里。恢复时先重放这些缓冲数据。
| 维度 | Aligned Checkpoint | Unaligned Checkpoint |
|---|---|---|
| Barrier等待 | 等所有流对齐 | 不等，直接Checkpoint |
| 延迟影响 | 可能很大（如果某个流慢） | 小（不等待） |
| Checkpoint大小 | 只保存状态 | 状态 + 缓冲数据（更大） |
| 适用场景 | 数据量不大、反压不严重 | 反压严重、对延迟敏感 |
端到端 Exactly-Once：两阶段提交
引擎内部的Exactly-Once只保证Flink内部状态不丢不重。但如果Sink写入外部系统（如Kafka、数据库），还需要Sink支持事务。
Flink的Kafka Sink用两阶段提交（2PC）实现端到端 Exactly-Once：
- 预提交（Pre-commit）： Checkpoint时，Sink把数据写入Kafka但不提交事务
- 提交（Commit）： Checkpoint成功后，JobManager通知Sink提交事务，数据对下游可见
- 回滚（Abort）： Checkpoint失败时，回滚未提交的事务，数据不会被下游看到
// Kafka Sink 开启 Exactly-Once
KafkaSink.<String>builder()
.setBootstrapServers("kafka:9092")
.setRecordSerializer(KafkaRecordSerializationSchema.builder()
.setTopic("output-topic")
.setValueSerializationSchema(new SimpleStringSchema())
.build())
.setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 开启Exactly-Once
.setTransactionalIdPrefix("my-app") // 事务ID前缀，必须设置
.build();
// 注意：Kafka的transaction.max.timeout.ms要大于Checkpoint间隔
// 否则事务超时会被自动回滚！实战经验：Kafka事务超时的坑
我在元宇宙社交App第一次配置Kafka Exactly-Once Sink时，经常看到数据丢失。排查后发现：Kafka Broker的transaction.max.timeout.ms默认是15分钟，但我们的Checkpoint间隔是10分钟、超时是15分钟。当Checkpoint慢的时候，事务在Kafka侧已经超时被回滚了，但Flink还不知道。
解决： 把Kafka的transaction.max.timeout.ms调到大于Checkpoint超时时间的两倍。比如Checkpoint超时15分钟，Kafka设成30分钟。