62.4 Exactly-once 语义
一句话
流处理的 exactly-once 不是「消息只来一次」，而是状态、输入位点、输出提交三者在同一个检查点边界上原子生效；broker 的事务只覆盖其中一段，端到端要靠 sink 的 2PC 或幂等写补齐。
端到端四段，缺一段都不算
| 段 | 要保证什么 | 失败后果 | 谁负责 |
|---|---|---|---|
| 输入位点 | 恢复后从「状态对应的位置」继续读 | 提前提交 → 丢；滞后提交 → 重 | source / checkpoint |
| 算子状态 | 状态与位点是同一个版本 | 状态旧、位点新 → 丢；反之 → 重 | checkpoint（62.3 状态管理与容错） |
| 输出写入 | 写入对外可见的时机可控 | 写早了 → 恢复后重复 | sink 事务/幂等 |
| 提交 | 输出提交与 checkpoint 完成对齐 | commit 早于 checkpoint → 重复 | 2PC 协议 |
🔑 消息队列的「事务消息 / 幂等生产」只解决日志里不重复；它管不到「消费后写数据库」这一段。端到端 exactly-once 的主场在 sink，不在 broker。消息侧的语义拆解见 43.2 投递语义 (at-most / at-least / exactly-once)。
2PC Sink 的协议顺序
checkpoint 的 barrier 对齐与状态快照见 62.3 状态管理与容错；实时数仓的 OLAP upsert 落地见 62.6 实时数仓。
begin tx ──> 处理事件，写 staged（对外不可见）
│
checkpoint barrier
│
preCommit（flush，准备提交）
│
checkpoint 完成（状态 + 位点 + tx id 落盘）
│
commit（幂等；失败可重试）
恢复时的规则：
| 恢复时看到 | 动作 |
|---|---|
| 事务已提交 | 跳过，不重放该批 |
| 事务已 prepare、checkpoint 未完成 | abort，丢弃 staged，从旧位点重放 |
| 事务已 prepare、checkpoint 已完成 | 重新 commit（幂等），或确认已提交后跳过 |
| 没有事务记录 | 按 at-least-once 处理，靠 sink 幂等兜底 |
关键顺序只有一条：commit 必须发生在 checkpoint 完成之后。 反过来（先 commit 再完成 checkpoint），崩溃恢复会回到上一个 checkpoint，把已经提交的数据再写一遍。
代码：四种崩溃/提交顺序
package main
import "fmt"
const (
total = 20 // 20 个事件，每个 +1
batch = 5 // 每 5 个事件一个 checkpoint
)
// atLeastOnce：处理一条写一条；崩溃后从上一个 checkpoint 的位点重放。
func atLeastOnce(crashAt int) (sink, dup int) {
sink = crashAt
lastCP := crashAt / batch * batch
for i := lastCP + 1; i <= total; i++ {
sink++
}
return sink, sink - total
}
// twoPhase：prepare -> checkpoint 完成 -> commit；恢复时未提交的事务 abort。
func twoPhase(crashMode string) (sink int) {
for i := 1; i <= total; i++ {
if i%batch != 0 {
continue
}
// 到这里：本批 5 条已写入 staged，准备 pre-commit
if i == 10 {
switch crashMode {
case "beforeCommit":
// 恢复点：最后一个完成的 checkpoint 是 5，staged 丢弃，重放 6..20
return replayTwoPhase(6, total, 5)
case "afterCommitEarly":
// 协议错误：checkpoint 还没完成就 commit。恢复点仍是 5，
// 但 6..10 已经在 sink 里 -> 重放会重复。
return replayTwoPhase(6, total, 10)
case "afterCommit":
// 正确顺序：checkpoint 10 已完成，commit 也已生效，
// 恢复后从 11 继续，不重放 6..10。
return 10 + (total - 10)
}
}
// 正常路径：checkpoint 完成，然后 commit 本批
sink += batch
}
return sink
}
func replayTwoPhase(from, to, base int) int {
sink := base
for b := from; b <= to; b += batch {
sink += batch
}
return sink
}
func main() {
fmt.Println("=== 同一份 20 条 +1 事件，四种崩溃/提交顺序 ===")
fmt.Printf("%-30s %-10s %-8s %-8s %s\n", "scenario", "sink", "expect", "dup", "lost")
aSink, aDup := atLeastOnce(7)
fmt.Printf("%-30s %-10d %-8d %-8d %d\n", "at-least-once crash@7", aSink, total, aDup, 0)
bSink := twoPhase("beforeCommit")
fmt.Printf("%-30s %-10d %-8d %-8d %d\n", "2PC crash before commit", bSink, total, 0, 0)
cSink := twoPhase("afterCommit")
fmt.Printf("%-30s %-10d %-8d %-8d %d\n", "2PC crash after commit", cSink, total, 0, 0)
dSink := twoPhase("afterCommitEarly")
fmt.Printf("%-30s %-10d %-8d %-8d %d\n", "2PC commit before CP done", dSink, total, dSink-total, 0)
fmt.Println("\n=== 协议顺序 ===")
fmt.Println("1) 处理事件 -> 写入 sink 的 staged（对外不可见）")
fmt.Println("2) checkpoint 快照 = 算子状态 + 未提交事务 ID + 输入位点")
fmt.Println("3) checkpoint 完成后，才 commit 事务（提交必须幂等）")
fmt.Println("4) 恢复时：已提交的事务跳过；未提交的事务 abort 并从位点重放")
fmt.Println("\n结论：2PC 的 exactly-once 不取决于“有没有事务”，而取决于")
fmt.Println(" “commit 是否严格发生在 checkpoint 完成之后”，以及恢复时能否识别已提交事务。")
}
本机实测（macOS，Apple M1，Go 1.27.1）：
=== 同一份 20 条 +1 事件，四种崩溃/提交顺序 ===
scenario sink expect dup lost
at-least-once crash@7 22 20 2 0
2PC crash before commit 20 20 0 0
2PC crash after commit 20 20 0 0
2PC commit before CP done 25 20 5 0
=== 协议顺序 ===
1) 处理事件 -> 写入 sink 的 staged（对外不可见）
2) checkpoint 快照 = 算子状态 + 未提交事务 ID + 输入位点
3) checkpoint 完成后，才 commit 事务（提交必须幂等）
4) 恢复时：已提交的事务跳过；未提交的事务 abort 并从位点重放
结论：2PC 的 exactly-once 不取决于“有没有事务”，而取决于
“commit 是否严格发生在 checkpoint 完成之后”，以及恢复时能否识别已提交事务。
读数：
| 场景 | sink 结果 | 重复 | 病因 |
|---|---|---|---|
| at-least-once，第 7 条崩溃 | 22 | +2 | 6、7 已写入，恢复回 checkpoint 5，重放 6..20 |
| 2PC，prepare 后、commit 前崩溃 | 20 | 0 | staged 未提交，恢复 abort，重放一次 |
| 2PC，commit 后崩溃（checkpoint 已完成） | 20 | 0 | checkpoint 10 含 tx；恢复识别已提交，跳过 |
| 2PC，commit 早于 checkpoint 完成 | 25 | +5 | 恢复回 checkpoint 5，6..10 已提交又被重放 |
🔑 第 4 行是本节最值钱的一行：有事务 ≠ exactly-once。事务只是让写入「可提交/可回滚」，提交时机错了，恢复一样重复。判断一个 sink 是否真的 exactly-once，要问三句话：commit 在 checkpoint 完成之后吗？tx id 在 checkpoint 里吗？恢复时能识别「已提交」吗？
2PC 与幂等写：两条路，不是一回事
| 2PC Sink | 幂等写（upsert / 去重键） | |
|---|---|---|
| 前提 | sink 支持事务（Kafka、支持 XA/事务的数据库） | sink 支持按业务键覆盖/唯一约束 |
| 能保证 | 任意写入（append、聚合、更新） | 只对「同一 key 覆盖」或「同一事件 ID 去重」有效 |
| 延迟 | 高（事务提交、两阶段） | 低（一次写入） |
| 恢复 | 依赖 tx id 与 checkpoint | 依赖去重键保留期 |
| 典型 | Kafka 事务、Flink TwoPhaseCommitSinkFunction | 订单表 order_id 唯一键、Redis SETNX |
⚠️ 幂等写不是万能的：
count += 1这种「增量」写入，重放时再+1就错了；必须改成「按事件 ID 去重后 set 最终值」或「用事务」。很多团队把「幂等」理解成「重复执行不出错」，但增量操作天然不幂等。
成本：exactly-once 不是免费的
| 成本 | 表现 | 缓解 |
|---|---|---|
| 延迟 | 事务提交要等 checkpoint 完成 | 缩短 checkpoint interval，但别短到跑不完 |
| 吞吐 | 事务标记、两阶段提交、状态版本 | 批量提交、减少小事务 |
| 事务超时 | checkpoint 太慢，事务超时被 broker abort | 对齐 Flink transaction.timeout.ms 与 broker transaction.max.timeout.ms |
| 状态 | 未提交事务要留在 checkpoint 里 | 控制并发事务数、及时 commit/abort |
| 外部系统 | 不是所有 sink 都支持 2PC | 不支持就退到 at-least-once + 幂等/去重 |
Kafka 侧的经典坑：Flink 的 transaction.timeout.ms 默认是 1 小时，而 Kafka broker 的 transaction.max.timeout.ms 默认是 15 分钟。不对齐时，checkpoint 稍慢就会触发 InvalidTxnTimeoutException，作业反复重启。
生产落地：指名 → 可抄 → 可验 → 可退
指名：Flink + Kafka 用 Kafka 事务 sink（EXACTLY_ONCE + TwoPhaseCommitSinkFunction / 新版 SinkV2 的 Committer）；数据库 sink 用支持事务的 JDBC 或幂等 upsert；Kafka Streams 用 processing.guarantee=exactly_once_v2。
可抄（Flink 1.17+ + Kafka，未验证：本机未安装 Flink/Kafka）：
# Flink
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.interval: 10s
execution.checkpointing.timeout: 10min
execution.checkpointing.min-pause: 5s
# Kafka producer（Flink KafkaSink 内部配置）
enable.idempotence=true
transactional.id=flink-gmv-1
transaction.timeout.ms=900000 # 必须 <= broker 的 transaction.max.timeout.ms
# 下游消费者：只读已提交，避免看到未提交事务的数据
isolation.level=read_committed
可验：本节程序输出 22/20/20/25，直接看「commit 时机」对结果的影响；Kafka 侧用 kafka-console-consumer --isolation-level read_committed 验证未提交数据不可见；Flink UI 看 checkpoint 是否包含 pending transaction。可退：sink 不支持事务时，切回 at-least-once + 幂等写（明确记录「允许重复、不允许丢」）；事务超时导致失败时，先调大 broker transaction.max.timeout.ms 或缩短 checkpoint，不要直接关 exactly-once。
反模式
| 反模式 | 后果 |
|---|---|
| 认为「开了事务消息 = 端到端 exactly-once」 | 写库/调接口仍重复 |
| commit 早于 checkpoint 完成 | 恢复回旧 checkpoint，已提交数据再写一遍（本节 +5） |
| checkpoint 不存 tx id | 恢复时不知道事务状态，只能重放或丢 |
| commit 不幂等 | 恢复后重复 commit，sink 出现重复数据 |
| 事务超时 > broker 允许上限 | 作业反复 InvalidTxnTimeoutException |
| 用增量写做「幂等」 | +1 重放变 +2，越补越多 |
sink 不支持事务却开 EXACTLY_ONCE | 配置看似生效，实际只有 at-least-once |
| 长 checkpoint 跨过事务超时 | 事务被 abort，数据丢失或重复 |
自检
Q1 端到端 exactly-once 需要哪四段？broker 事务覆盖哪一段？
▸ 答案：输入位点、算子状态、输出写入、提交四段。broker 事务/幂等生产只覆盖「日志里不重复」这一段（生产端 + 存储端），不覆盖「消费后写 sink」的输出段。输出段要么由 sink 的 2PC 事务保证，要么由幂等写/去重键保证。
Q2 2PC sink 的正常顺序是什么？为什么 commit 必须在 checkpoint 完成之后？
▸ 答案：处理事件写 staged → barrier → preCommit（flush）→ checkpoint 完成（状态 + 位点 + tx id 落盘）→ commit。commit 在 checkpoint 完成后，崩溃恢复时才能确定「这个事务对应的 checkpoint 已经持久化」；如果先 commit 再完成 checkpoint，恢复会回到上一个 checkpoint，把已提交的数据再写一遍。
Q3 本节「2PC commit before CP done」为什么多了 5 条？
▸ 答案：checkpoint 10 还没完成就 commit 了 6..10，sink 已有 10 条；崩溃后恢复到最后一个完成的 checkpoint 5，重放 6..20，其中 6..10 再写一次，所以 sink = 10 + 15 = 25，重复 5 条。根因不是事务，而是提交顺序。
Q4 at-least-once 场景为什么是 22 而不是 20？重复的 2 条从哪来？
▸ 答案：第 7 条崩溃时，sink 已经写了 1..7；恢复回 checkpoint 5，从 6 重放到 20，6、7 被写第二次。sink = 7 + 15 = 22，重复 2 条。这就是「处理完再提交」的代价：不丢，但会重。
Q5 什么场景用 2PC，什么场景用幂等写？
▸ 答案：sink 支持事务（Kafka、支持事务的数据库）且写入是 append/聚合/更新混合时，用 2PC；sink 只支持按 key 覆盖或有唯一约束，且写入可表达为 upsert 时，用幂等写。增量写（+1）必须先转成「按事件 ID 去重 + 设最终值」才能幂等，否则不能用幂等写替代 2PC。
Q6 Kafka 事务超时的经典坑是什么？
▸ 答案：Flink 的 transaction.timeout.ms 默认 1 小时，Kafka broker 的 transaction.max.timeout.ms 默认 15 分钟；不对齐时，checkpoint 稍慢就会超时，broker abort 事务，作业反复失败。生产上要让前者 ≤ 后者，并保证 checkpoint + commit 能在超时内完成。
Q7 checkpoint 里为什么要存「未提交事务 ID」？
▸ 答案：恢复时算子需要知道「上次 checkpoint 时有哪些事务已经 prepare 但还没 commit」。存了 tx id，才能对「checkpoint 已完成」的事务重新 commit（幂等），对「checkpoint 未完成」的事务 abort；不存就只能盲重放，导致重复或丢失。