下游对账的同学在群里甩了张截图同一笔订单在报表里出现了两次。诡异的是作业从上线到现在一次都没重启过Checkpoint 页面上一直显示成功状态大小也正常。查了半天才发现问题出在 Sink 端——Flink 内部的状态确实回滚得干干净净但数据早就写进外部系统了Checkpoint 根本管不着它。这种事在第一次接触Flink CheckPoint和两阶段提交协议Two-Phase Commit Protocol的人身上几乎必然发生一次。绝大多数教程会把 2PC 讲成Flink 帮你把事务管好了但真正上线之后你会发现能不能拿到端到端的精确一次取决于外部系统支不支持事务、checkpoint 间隔配得对不对、事务超时留没留够、恢复路径走的是 commit 还是 abort。这篇文章就把这套机制从里到外拆一遍包括我在实际项目里踩过的几个坑以及手写一个符合 2PC 语义的 Sink 时该注意什么。1. Checkpoint 只能救回 Flink 自己的状态救不回已经写出去的数据1.1 一次 Checkpoint 在集群里到底动了哪些东西先明确一点Flink 的 Checkpoint 是一份分布式快照它拍的是算子内部的状态不是数据本身。JobManager 里的 CheckpointCoordinator 按execution.checkpointing.interval周期性地往下游注入一个叫 barrier 的特殊标记barrier 随数据流一起流动算子必须等所有输入通道的 barrier 都到齐也就是常说的对齐才会对自己当前的状态做一次快照然后把 barrier 继续往下游转发。快照做完之后并不会立刻算成功。算子把状态异步上传到 checkpoint storageHDFS、对象存储或者 JobManager 内存全部上传完成、JobManager 收到所有算子的确认之后这次 checkpoint 才被标记为 completed。这个完成的时间点非常关键后面讲 2PC 的 commit 时机全靠它。所以 Checkpoint 能保证的是作业失败重启后每个算子的状态能回到某个一致的时间点。比如你算累加值重启后不会重复累加比如 Kafka Source 记住的 offset重启后会从正确的位置重新消费。这条链路是闭合的因为消费位置也属于状态。1.2 外部系统的写入为什么是漏网之鱼问题出在算子往外写数据的那一步。假设你的 Sink 是往 MySQL 里插一条记录数据插进去了MySQL 那边事务已提交、连接已释放。这时候作业挂了重启后从上一个成功的 checkpoint 恢复那段已经插过的数据会被重新算一遍、重新插一遍——外部系统完全不知道 Flink 在回滚它只认自己收到的那些写入。这就是端到端精确一次的核心矛盾Flink 的状态可以回滚外部系统的副作用不能回滚。解决思路只有三条路让外部系统支持事务把写入放进一个可以先悬着、再统一提交的事务里——这就是两阶段提交协议要做的事。让写入变成幂等的重复写多少次结果都一样比如按主键 upsert、按 rowkey 覆盖、按 doc id 覆盖。先写日志WAL等 checkpoint 成功后再把日志里的内容搬到目标系统。Flink 选择了第一条路作为默认方案把它封装成了TwoPhaseCommitSinkFunction以及后来的TwoPhaseCommittingSink。理解这套封装之前得先把经典的 2PC 想明白。2. 把 Flink Checkpoint 当成分布式事务的协调者2.1 经典 2PC 的三个角色和两个阶段教科书里的两阶段提交有三个角色协调者Coordinator、参与者Participant、以及一个持久化的日志。两个阶段分别是准备阶段协调者问所有参与者你能提交吗参与者把数据写好、锁住资源、把 undo/redo 信息刷到日志然后回答可以或者不行。这个回答一旦给出就不能反悔。提交阶段如果所有人都说可以协调者决定提交通知所有人执行提交只要有一个人说不行就通知所有人回滚。这套协议最大的代价是阻塞——参与者从给出可以到收到最终决定之间资源一直被锁着。这也是为什么传统数据库里的分布式事务在高并发下表现一般。2.2 Flink 版本的两阶段提交只保留了 prepare 和 commitFlink 的做法很巧妙它把Checkpoint 完成这件事本身当成了协调者的投票结果省掉了询问能不能提交这一轮交互。具体映射关系是这样的一个 checkpoint 周期内所有 Sink 子任务写了但还没最终确认的数据处于预提交状态当这个 checkpoint 被 JobManager 标记为 completed 时就等于协调者宣布这一批全部通过于是通知所有 Sink 执行真正的提交如果这个 checkpoint 中途失败或被取消就等于宣布作废所有 Sink 回滚自己那一批数据。这里有个容易被忽略的细节Checkpoint 的 barrier 把数据流天然切成了两段。barrier 之前到达 Sink 的数据属于当前事务barrier 之后到达的数据属于下一个事务。因为 barrier 对齐保证了同一个 checkpoint 的所有数据都在同一条分界线两侧事务边界和 checkpoint 边界严格重合这就避免了一部分数据算老事务、一部分算新事务的错乱。开启非对齐 Checkpointunaligned checkpoint之后barrier 不再等待对齐部分在途数据会被塞进状态里一起快照但事务边界的语义没有变仍然是按 barrier 划分。另外要澄清一个常见误解Flink 里的 2PC不是所有人投票、少数服从多数那种原子提交协议。它没有投票环节也没有真正意义上的全局锁。它的本质是本地事务先挂起等 checkpoint 成功后再统一提交是靠 Checkpoint 的原子性和持久化来兜底的。这个差别决定了它的一个致命弱点——只要 checkpoint 一直不成功事务就一直悬着外部系统里的锁和资源就一直不释放。这也是后面讲事务超时那个坑的根源。3. TwoPhaseCommitSinkFunction 的钩子函数到底按什么顺序被调用3.1 beginTransaction / invoke / preCommit 的日常循环如果你看过老版本的 Flink Sink 代码一定见过这几个方法。这里按实际调用顺序捋一遍顺序搞错了逻辑必然写错。作业刚启动或者从状态恢复时Sink 会调用一次beginTransaction()拿到一个事务句柄比如一个 Kafka 的 producer、一个 XA 连接。这个句柄会被保存在一个成员变量里同时也要作为状态的一部分被 checkpoint。接下来是常规写入循环每来一条数据invoke(txn, value, context)被调用把数据写进当前事务的缓冲区。注意这时候数据只是写进去了对下游还不可见——Kafka 事务里的消息在 commit 之前对read_committed的消费者是看不到的。当某个 checkpoint 的 barrier 到达 Sink 时preCommit(txn)被调用。这个方法的语义是把当前事务封口让它不再接受新数据并做好最终提交的准备。对于 Kafka这里通常就是producer.flush()对于文件系统 Sink这里就是把临时文件从.inprogress状态准备好转正。preCommit必须返回一个可以序列化的对象因为它要被塞进状态里存起来。preCommit返回之后Sink 立刻调用新的beginTransaction()开启下一个事务后续数据写进新事务里。老事务的句柄则被放进一个MapLong, ListTXN结构里key 是 checkpoint id——这个结构就是pendingCommitTransactions它会随 checkpoint 一起被持久化。3.2 checkpoint 完成通知到来时的 commit 分支真正的提交发生在notifyCheckpointComplete(checkpointId)回调里。JobManager 确认某个 checkpoint 完成后会向所有算子发送这个通知。Sink 收到之后从pendingCommitTransactions里取出这个 checkpoint id 对应的所有事务句柄逐个调用commit(txn)。这里有几个非常关键的工程细节第一commit必须是幂等的。因为通知可能重复送达比如某个算子的通知失败重试也可能在恢复后被recoverAndCommit再次触发。你重复 commit 同一个事务不能产生额外副作用。第二commit的开销要控制住。如果一次 checkpoint 里积累了几十万条消息commit 那一下的耗时可能很长这段时间会阻塞 Task 的主线程mailbox导致后续数据积压。实际项目里我一般会把 checkpoint 间隔和单批数据量一起调避免单个事务过大。第三老版本TwoPhaseCommitSinkFunction内部用一个单线程 executor 串行化 commit 操作并且在 checkpoint 和 commit 之间通过 mailbox 做同步防止出现新事务已经开了、老事务还在提交的混乱。你写自定义实现时如果也用了异步线程一定要自己保证这个顺序。第四从 Flink 1.15 开始TwoPhaseCommitSinkFunction已经被标记为废弃官方推的是 Sink V2 那套接口——TwoPhaseCommittingSink配一个独立的Committer组件把写数据和提交事务拆到两个算子Writer 和 Committer里。拆分的好处是提交不再占着写数据的主线程Committer 可以独立并行度。理解老接口是理解新接口的必经之路因为钩子的语义是一一对应的。3.3 失败恢复路径recoverAndCommit 与 recoverAndAbort作业挂掉重启时Sink 从最近一次成功的 checkpoint 里读出pendingCommitTransactions然后分两种情况处理如果某个 checkpoint 已经被标记为完成但commit还没执行完比如 JobManager 在发出通知后立刻宕机恢复时会调用recoverAndCommit(txn)把这些事务补交上。这是保证不丢的关键一环。如果某个 checkpoint 没有完成状态里存在但 JobManager 那边没有 completed 记录恢复时会调用recoverAndAbort(txn)把这些事务回滚掉。这是保证不重的关键一环。判断依据就是 checkpoint id 和已完成 checkpoint 列表的比对。这个逻辑在TwoPhaseCommitSinkFunction.initializeState里实现具体做法是把状态里的事务按 checkpoint id 排序落在已完成区间内的走 commit落在未完成区间内的走 abort。这里有个实战经验如果 checkpoint 存储本身不可靠这套恢复逻辑就废了。比如你把 checkpoint 存在 JobManager 内存里JobManagerCheckpointStorageJobManager 一挂状态和事务记录全丢重启后既不知道该 commit 谁也不知道该 abort 谁外部系统里就会留下一堆永远不结束的悬挂事务。生产环境老老实实配 HDFS 或者对象存储这不是最好有是必须有。4. Kafka Sink 是最标准的参考实现也是最好的教材4.1 transactional.id 的生成规则决定了事务不会串味Kafka 的事务是靠transactional.id做隔离的。这个 ID 是生产者实例的唯一身份Kafka 用它来 fencing隔离旧的、可能还在活动的生产者实例。Flink 的 Kafka Sink 不会让你直接指定完整的transactional.id你只能配一个前缀setTransactionalIdPrefix。最终生成的 ID 是前缀 子任务索引 自增序号的组合而且这个自增序号本身也保存在状态里NextTransactionalIdHint随 checkpoint 一起持久化。为什么要这么设计因为如果每次恢复都用同一个transactional.id新的 producer 会去恢复那个 ID 下未完成的事务这本身没问题但如果transactional.id重复且旧实例还活着比如网络分区导致的脑裂Kafka 就会把其中一个 fencing 掉作业直接报ProducerFencedException。加上自增序号之后每次新事务拿到的 ID 都是全新的从根上避免了复用冲突。这里有个我踩过的坑曾经为了图省事多个作业共用同一个transactionalIdPrefix。结果两个作业的分区分配一重叠就开始互相 fencing日志里刷屏一样报错。每个启用了事务的 Sink 必须有自己的前缀前缀里最好带上作业名和版本号。4.2 事务超时是压垮线上作业的第一杀手这是我认为整个 2PC 链路里最容易出事的地方。Kafka 生产者有两个超时参数客户端侧的transaction.timeout.msbroker 侧的transaction.max.timeout.ms。客户端提交的事务超时时间如果超过 broker 侧的上限broker 会直接拒绝这个事务。而 Flink 的 Kafka 连接器默认会把客户端超时设成 1 小时Kafka broker 默认上限通常是 15 分钟——开箱即用的情况下这两者是打架的你必须先把 broker 侧的上限调大否则启用 exactly-once 的第一天就会挂。更隐蔽的是另一面如果事务超时时间设得太短比如按默认的 60 秒而你的 checkpoint 间隔是 3 分钟或者作业正在做一次耗时很长的全量恢复那么在事务提交之前broker 就会认为这个事务超时了主动把它中止掉。生产者下一次操作时收到InvalidTxnStateException或者直接进入致命错误状态整个作业挂掉且无法自动恢复。我在几个项目里总结出来的配置规则大致是这样的配置项建议值原因单次 checkpoint 间隔1 到 3 分钟太短会让事务过小、频繁提交太长会让恢复时间变长最大并发 checkpoint 数1并发 checkpoint 会拉长事务悬挂时间除非你确实需要Kafka 客户端transaction.timeout.ms≥ checkpoint 间隔 × (并发数 1) × 2给 checkpoint 和数据积压留足余量broker 端transaction.max.timeout.ms≥ 客户端设的值否则事务在提交前就被 broker 拒绝作业最大预期恢复时间必须小于事务超时恢复期间事务一直悬着超时就会被中止特别提醒作业最大预期恢复时间这一项经常被忽略。作业重启后从 checkpoint 恢复大状态可能要好几分钟这段重启期间旧事务是没有提交的。如果事务超时时间只够覆盖正常运行的 checkpoint 间隔一次意外重启就能让所有悬挂事务过期。4.3 消费端必须打开 read_committed这一条很多人会忘。Kafka 事务提交前消息对消费者是可见但不可读的。默认的isolation.level是read_uncommitted消费者会看到那些还没提交的消息一旦事务被 abort这些消息就凭空消失了或者更糟——你看到了两遍。所以下游如果也是 Flink 作业必须在 Kafka Source 上设置isolation.level read_committed。这个参数一开消费者只会读到已经提交的事务里的消息配合 Kafka 的 Last Stable Offset 机制未提交事务之后的消息也会被暂时挡住直到事务尘埃落定。顺带说一句Kafka 的事务提交消息本身也是写进 topic 的只是它是一个控制批次不会被普通消费者读到在read_committed模式下。这也是为什么 Kafka 的事务开销不算特别小高频提交事务对吞吐影响是实打实的。5. 自己写一个 2PC SinkXA 为例的完整骨架与踩点5.1 状态里能放什么、绝对不能放什么很多人在写自定义 Sink 时第一反应是把数据库连接放进事务句柄里然后在状态里存这个连接对象。这件事一定会失败。原因很简单TwoPhaseCommitSinkFunction的第二个泛型参数TXN必须是可序列化的因为pendingCommitTransactions会随 checkpoint 一起写进状态存储。而java.sql.Connection是不可序列化的KafkaProducer同样不可序列化。正确的做法是存重建事务所需的最小信息——比如 XA 事务的Xid包含 formatId、全局事务 ID、分支限定符三个字段都是基本类型恢复时用这些信息重新构造 XAResource再执行 prepare 之后的提交或回滚。public class MyXid implements Xid, Serializable { private final int formatId; private final byte[] globalTransactionId; private final byte[] branchQualifier; public MyXid(int formatId, byte[] gtrid, byte[] bqual) { this.formatId formatId; this.globalTransactionId gtrid; this.branchQualifier bqual; } Override public int getFormatId() { return formatId; } Override public byte[] getGlobalTransactionId() { return globalTransactionId; } Override public byte[] getBranchQualifier() { return branchQualifier; } }序列化器也要换成 Kryo 或者自定义的TypeSerializer因为 Flink 默认的序列化器对自定义类处理起来不够好用。老接口的构造方法大概是这个样子public class XaTwoPhaseSink extends TwoPhaseCommitSinkFunctionRow, MyTxnHandle, Void { public XaTwoPhaseSink() { super( new KryoSerializer(MyTxnHandle.class, new ExecutionConfig()), VoidSerializer.INSTANCE ); } // ... }5.2 preCommit 里到底该做哪一步这是整个实现里最需要想清楚的地方。XA 协议本身就提供了prepare阶段看上去和preCommit完美对应直接调xaResource.prepare(xid)就行了。但这里面有个陷阱。prepare之后数据库那边的分支事务处于准备好了、等待协调者指令的状态锁一直持有。如果prepare卡住很久比如连接池打满、数据库抖动整个 Sink 的写数据主线程就被阻塞了。所以实践中我更倾向于在preCommit里做轻量的收尾动作——把缓冲区 flush 到中间的临时表或者临时文件真正耗时的prepare和commit放到 commit 阶段去做并且加超时控制。Override protected void preCommit(MyTxnHandle handle) throws Exception { // 只做封口和 flush不做重量级操作 handle.flushBuffer(); } Override protected void commit(MyTxnHandle handle) { try { XAResource resource handle.openResource(); resource.prepare(handle.getXid()); resource.commit(handle.getXid(), false); handle.closeQuietly(); } catch (Exception e) { // 关键commit 失败要让 checkpoint 感知到 throw new RuntimeException(commit failed for xid handle.getXid(), e); } }注意commit里抛异常的处理。如果提交失败却不抛出作业会认为一切正常数据就真丢了。但抛出异常会导致作业重启重启后recoverAndCommit还会再试一次——所以整个 commit 路径必须幂等数据库侧的 XA 事务重复提交会返回XAER_NOTA事务不存在这个错误码要当成已经提交过来处理不能当成失败。5.3 幂等兜底为什么 commit 必须可以重复执行上面反复提到幂等这里集中说一下为什么。notifyCheckpointComplete这个通知不保证一定送达。JobManager 在发出通知的过程中宕机这个通知就丢了。听起来很吓人但实际上不影响正确性——因为那些没被通知到的事务还在状态里下次恢复时recoverAndCommit会补上。但如果通知重复送达呢同样可能发生。Flink 在某些版本里会对通知做重试加上恢复路径的重放同一个事务可能被 commit 两次。如果你的commit实现是发一条 MQ 消息给下游这种有副作用的动作重复执行就会出问题。private static final SetString COMMITTED ConcurrentHashMap.newKeySet(); Override protected void commit(MyTxnHandle handle) { String key handle.getXid().toString(); if (!COMMITTED.add(key)) { return; // 已经提交过了直接跳过 } // ... 真正提交 }用内存 Set 只是个示意生产环境里应该依赖外部系统自身的幂等能力比如数据库的唯一约束、Kafka 事务的重复提交返回XAER_NOTA、对象存储的覆盖写。6. 文档里翻不到的那几个坑从 checkpoint 存储到并发检查点6.1 Checkpoint 存储选内存等于把事务状态押给了 JobManager前面提过一次这里再强调一遍因为它的后果比想象中严重。JobManagerCheckpointStorage把 checkpoint 数据存在 JobManager 的堆内存里好处是快坏处是一旦 JobManager 进程挂掉所有已完成和未完成的 checkpoint 记录一起消失。对于 2PC 来说这意味着悬挂在外部系统里的事务既不会被提交也不会被回滚会一直占用锁和资源直到超过事务超时被强制中止作业重启后无法从任何 checkpoint 恢复只能从零开始那些悬挂事务就永远成了孤儿。生产环境必须用文件系统类存储。HDFS 是很多公司的默认选择但如果你在云上对象存储S3、OSS 这类也很常见。这里有个额外的坑对象存储的 rename 不是原子操作它是 copy delete。文件系统类 Sink 通常靠 rename 来实现两阶段提交写.inprogress文件 → commit 时转正在对象存储上这个转正过程会有短暂的两个文件都存在或者中间态可被读到的窗口。如果你的下游是直接扫目录的批处理任务建议配合_SUCCESS标记文件一起判断。6.2 并发 checkpoint 数与事务超时的乘法关系execution.checkpointing.max-concurrent-checkpoints默认是 1很多人不知道这个值可以大于 1也不知道它在 2PC 场景下的影响。设成 1 的时候任意时刻只有一个 checkpoint 在跑事务悬挂的时间大约是一个 checkpoint 的完整耗时。设成 2 之后前一个 checkpoint 还没完成后一个已经开始了事务悬挂的时间可能翻倍而且pendingCommitTransactions里会同时存在多个 checkpoint 的事务。如果这时候事务超时时间没跟着调大就会出现前一个事务被 broker 中止的故障。我的建议是除非你的 checkpoint 特别慢、必须用并发来掩盖否则保持 1。真要用并发事务超时时间至少按并发数等比放大。另外还有个相关的坑checkpoint 的对齐时间。如果某个算子的输入通道数据倾斜严重barrier 对齐可能要等很久这段时间当前事务一直处于写了但没 preCommit的状态。开启非对齐 checkpoint 能缓解这个问题因为 barrier 不再等待数据对齐但代价是 checkpoint 体积会变大在途数据也要快照。6.3 JDBC 类 Sink 为什么天然做不到端到端精确一次热词里常出现Flink 的 JDBC 连接器异常这里面相当一部分根因就跟事务语义有关。Flink 官方的 JDBC Sink 走的是批量攒批 预编译语句的路子它不是事务性的。你可以说它在某些数据库驱动下支持autoCommitfalse的批量提交但这个提交和 Flink 的 checkpoint 没有任何绑定关系——它可能在 checkpoint 完成之前就把数据提交了也可能在 checkpoint 失败后数据已经落库。那怎么办两条路第一条是幂等写入。给目标表建立唯一键SQL 改成INSERT ... ON DUPLICATE KEY UPDATEMySQL或者ON CONFLICT DO UPDATEPostgreSQL重复写的结果和写一次一样。这条路简单直接绝大多数场景够用代价是吞吐会打折因为每条记录都要走一次唯一性检查。第二条是自己在 JDBC Sink 之上实现 2PC用 XA 连接。代价是复杂度和延迟都会显著上升而且不是所有数据库的 XA 实现都足够可靠。如果你用的是 Doris、StarRocks 这类分析型数据库的连接器情况又不一样。它们一般提供自己的一套两阶段导入机制比如先写临时标签再统一生效需要看对应连接器版本文档里对 exactly-once 的支持程度。不同连接器对事务语义的支持差异非常大用了 Flink 就自动精确一次是个危险的假设接手一个新连接器时第一件事就是查它的语义保证级别。6.4 notifyCheckpointComplete 不保证送达但这不影响正确性这一点单独拿出来说是因为它经常被人误读成notifyCheckpointComplete 不可靠所以 2PC 不可靠。实际上这两件事是分开的通知只是加速提交的优化路径真正的正确性保障在状态里。只要pendingCommitTransactions被正确持久化哪怕所有通知都丢了作业下次重启时也会通过recoverAndCommit把该提交的事务补上。数据只会晚一点对外可见不会丢。真正需要担心的反而是另一面通知送达了但 commit 失败了。这种情况下事务还挂在外部系统里如果 checkpoint 已经被 subsumed被后续的 checkpoint 清理掉恢复时就找不到这个事务了。所以 commit 失败一定要抛出异常让作业重启绝不能吞掉。7. 什么时候该放弃 2PC幂等写入和 WAL 的更划算选择7.1 幂等写入的成本核算2PC 不是免费的午餐。它带来的额外开销包括每个 checkpoint 周期一次批量提交的网络往返、外部系统的锁竞争、事务日志的持久化开销以及事务超时带来的运维复杂度。如果你的 Sink 天然支持幂等按主键覆盖、按唯一约束去重、版本号覆盖那用幂等方案往往更划算。整个链路变成 at-least-once但结果等价于 exactly-once。实现上只需要在算子里维护一个可以重放的输出流写好 SQL 或者 API 调用即可。判断标准很实用问自己一句同一条数据写两次结果会不会不一样如果不会就上幂等如果会比如累加、计数、发消息通知才需要考虑 2PC 或者别的补偿机制。方案语义对下游要求延迟开销适用场景两阶段提交精确一次必须支持事务高受 checkpoint 间隔制约金融流水、订单主表、不能重复的通知幂等写入至少一次但结果等价有唯一键或覆盖写能力低维表同步、日志汇聚、指标明细WAL 中转精确一次有一个可靠的中间存储中多一次搬运目标系统不支持事务但能接受最终一致7.2 WAL 方案的适用边界WAL 的思路是Sink 不直接写目标系统先把数据写到一个高可靠的中间存储通常是 HDFS 或者对象存储上的文件等 checkpoint 成功后再由一个独立的搬运任务把文件内容导入目标系统。导入过程靠文件的命名或者目录结构区分已提交和未提交导入任务只处理已提交的文件并记录自己的处理进度。这条路的典型代表就是文件系统类 Sink。它在很多场景下比 2PC 更省心因为中间存储的能力是可控的不依赖目标系统的事务支持。代价是数据可见性有延迟至少一个 checkpoint 周期以及多维护一套搬运逻辑。我在一个日增量几十亿条的项目里最终选的就是 WAL 路线因为目标系统当时还不支持跨会话的长事务。回过头看这个选择的判断依据很简单先看目标系统能给什么再决定用哪套语义而不是反过来让业务去凑技术方案。最后分享两个我在实际运维里养成的习惯。第一个是给每个启用了事务语义的 Sink 单独建一张台账记录它的 checkpoint 间隔、事务超时、重启耗时上限任何一项改动都同步更新——这三个数是有乘法关系的改一个忘了另一个就容易出事。第二个是上线前手动做一次杀进程演练作业跑着的时候直接 kill 掉 TaskManager观察外部系统里有没有悬挂事务、重启后数据有没有重复或者丢失。这个演练花二十分钟比在生产上排查半天要划算得多。 
长河君 · 2026/9/18 13:21:18 · 阅读 1.2k Flink Checkpoint 与两阶段提交：端到端精确一次 Sink 实战
下游对账的同学在群里甩了张截图同一笔订单在报表里出现了两次。诡异的是作业从上线到现在一次都没重启过Checkpoint 页面上一直显示成功状态大小也正常。查了半天才发现问题出在 Sink 端——Flink 内部的状态确实回滚得干干净净但数据早就写进外部系统了Checkpoint 根本管不着它。这种事在第一次接触Flink CheckPoint和两阶段提交协议Two-Phase Commit Protocol的人身上几乎必然发生一次。绝大多数教程会把 2PC 讲成Flink 帮你把事务管好了但真正上线之后你会发现能不能拿到端到端的精确一次取决于外部系统支不支持事务、checkpoint 间隔配得对不对、事务超时留没留够、恢复路径走的是 commit 还是 abort。这篇文章就把这套机制从里到外拆一遍包括我在实际项目里踩过的几个坑以及手写一个符合 2PC 语义的 Sink 时该注意什么。1. Checkpoint 只能救回 Flink 自己的状态救不回已经写出去的数据1.1 一次 Checkpoint 在集群里到底动了哪些东西先明确一点Flink 的 Checkpoint 是一份分布式快照它拍的是算子内部的状态不是数据本身。JobManager 里的 CheckpointCoordinator 按execution.checkpointing.interval周期性地往下游注入一个叫 barrier 的特殊标记barrier 随数据流一起流动算子必须等所有输入通道的 barrier 都到齐也就是常说的对齐才会对自己当前的状态做一次快照然后把 barrier 继续往下游转发。快照做完之后并不会立刻算成功。算子把状态异步上传到 checkpoint storageHDFS、对象存储或者 JobManager 内存全部上传完成、JobManager 收到所有算子的确认之后这次 checkpoint 才被标记为 completed。这个完成的时间点非常关键后面讲 2PC 的 commit 时机全靠它。所以 Checkpoint 能保证的是作业失败重启后每个算子的状态能回到某个一致的时间点。比如你算累加值重启后不会重复累加比如 Kafka Source 记住的 offset重启后会从正确的位置重新消费。这条链路是闭合的因为消费位置也属于状态。1.2 外部系统的写入为什么是漏网之鱼问题出在算子往外写数据的那一步。假设你的 Sink 是往 MySQL 里插一条记录数据插进去了MySQL 那边事务已提交、连接已释放。这时候作业挂了重启后从上一个成功的 checkpoint 恢复那段已经插过的数据会被重新算一遍、重新插一遍——外部系统完全不知道 Flink 在回滚它只认自己收到的那些写入。这就是端到端精确一次的核心矛盾Flink 的状态可以回滚外部系统的副作用不能回滚。解决思路只有三条路让外部系统支持事务把写入放进一个可以先悬着、再统一提交的事务里——这就是两阶段提交协议要做的事。让写入变成幂等的重复写多少次结果都一样比如按主键 upsert、按 rowkey 覆盖、按 doc id 覆盖。先写日志WAL等 checkpoint 成功后再把日志里的内容搬到目标系统。Flink 选择了第一条路作为默认方案把它封装成了TwoPhaseCommitSinkFunction以及后来的TwoPhaseCommittingSink。理解这套封装之前得先把经典的 2PC 想明白。2. 把 Flink Checkpoint 当成分布式事务的协调者2.1 经典 2PC 的三个角色和两个阶段教科书里的两阶段提交有三个角色协调者Coordinator、参与者Participant、以及一个持久化的日志。两个阶段分别是准备阶段协调者问所有参与者你能提交吗参与者把数据写好、锁住资源、把 undo/redo 信息刷到日志然后回答可以或者不行。这个回答一旦给出就不能反悔。提交阶段如果所有人都说可以协调者决定提交通知所有人执行提交只要有一个人说不行就通知所有人回滚。这套协议最大的代价是阻塞——参与者从给出可以到收到最终决定之间资源一直被锁着。这也是为什么传统数据库里的分布式事务在高并发下表现一般。2.2 Flink 版本的两阶段提交只保留了 prepare 和 commitFlink 的做法很巧妙它把Checkpoint 完成这件事本身当成了协调者的投票结果省掉了询问能不能提交这一轮交互。具体映射关系是这样的一个 checkpoint 周期内所有 Sink 子任务写了但还没最终确认的数据处于预提交状态当这个 checkpoint 被 JobManager 标记为 completed 时就等于协调者宣布这一批全部通过于是通知所有 Sink 执行真正的提交如果这个 checkpoint 中途失败或被取消就等于宣布作废所有 Sink 回滚自己那一批数据。这里有个容易被忽略的细节Checkpoint 的 barrier 把数据流天然切成了两段。barrier 之前到达 Sink 的数据属于当前事务barrier 之后到达的数据属于下一个事务。因为 barrier 对齐保证了同一个 checkpoint 的所有数据都在同一条分界线两侧事务边界和 checkpoint 边界严格重合这就避免了一部分数据算老事务、一部分算新事务的错乱。开启非对齐 Checkpointunaligned checkpoint之后barrier 不再等待对齐部分在途数据会被塞进状态里一起快照但事务边界的语义没有变仍然是按 barrier 划分。另外要澄清一个常见误解Flink 里的 2PC不是所有人投票、少数服从多数那种原子提交协议。它没有投票环节也没有真正意义上的全局锁。它的本质是本地事务先挂起等 checkpoint 成功后再统一提交是靠 Checkpoint 的原子性和持久化来兜底的。这个差别决定了它的一个致命弱点——只要 checkpoint 一直不成功事务就一直悬着外部系统里的锁和资源就一直不释放。这也是后面讲事务超时那个坑的根源。3. TwoPhaseCommitSinkFunction 的钩子函数到底按什么顺序被调用3.1 beginTransaction / invoke / preCommit 的日常循环如果你看过老版本的 Flink Sink 代码一定见过这几个方法。这里按实际调用顺序捋一遍顺序搞错了逻辑必然写错。作业刚启动或者从状态恢复时Sink 会调用一次beginTransaction()拿到一个事务句柄比如一个 Kafka 的 producer、一个 XA 连接。这个句柄会被保存在一个成员变量里同时也要作为状态的一部分被 checkpoint。接下来是常规写入循环每来一条数据invoke(txn, value, context)被调用把数据写进当前事务的缓冲区。注意这时候数据只是写进去了对下游还不可见——Kafka 事务里的消息在 commit 之前对read_committed的消费者是看不到的。当某个 checkpoint 的 barrier 到达 Sink 时preCommit(txn)被调用。这个方法的语义是把当前事务封口让它不再接受新数据并做好最终提交的准备。对于 Kafka这里通常就是producer.flush()对于文件系统 Sink这里就是把临时文件从.inprogress状态准备好转正。preCommit必须返回一个可以序列化的对象因为它要被塞进状态里存起来。preCommit返回之后Sink 立刻调用新的beginTransaction()开启下一个事务后续数据写进新事务里。老事务的句柄则被放进一个MapLong, ListTXN结构里key 是 checkpoint id——这个结构就是pendingCommitTransactions它会随 checkpoint 一起被持久化。3.2 checkpoint 完成通知到来时的 commit 分支真正的提交发生在notifyCheckpointComplete(checkpointId)回调里。JobManager 确认某个 checkpoint 完成后会向所有算子发送这个通知。Sink 收到之后从pendingCommitTransactions里取出这个 checkpoint id 对应的所有事务句柄逐个调用commit(txn)。这里有几个非常关键的工程细节第一commit必须是幂等的。因为通知可能重复送达比如某个算子的通知失败重试也可能在恢复后被recoverAndCommit再次触发。你重复 commit 同一个事务不能产生额外副作用。第二commit的开销要控制住。如果一次 checkpoint 里积累了几十万条消息commit 那一下的耗时可能很长这段时间会阻塞 Task 的主线程mailbox导致后续数据积压。实际项目里我一般会把 checkpoint 间隔和单批数据量一起调避免单个事务过大。第三老版本TwoPhaseCommitSinkFunction内部用一个单线程 executor 串行化 commit 操作并且在 checkpoint 和 commit 之间通过 mailbox 做同步防止出现新事务已经开了、老事务还在提交的混乱。你写自定义实现时如果也用了异步线程一定要自己保证这个顺序。第四从 Flink 1.15 开始TwoPhaseCommitSinkFunction已经被标记为废弃官方推的是 Sink V2 那套接口——TwoPhaseCommittingSink配一个独立的Committer组件把写数据和提交事务拆到两个算子Writer 和 Committer里。拆分的好处是提交不再占着写数据的主线程Committer 可以独立并行度。理解老接口是理解新接口的必经之路因为钩子的语义是一一对应的。3.3 失败恢复路径recoverAndCommit 与 recoverAndAbort作业挂掉重启时Sink 从最近一次成功的 checkpoint 里读出pendingCommitTransactions然后分两种情况处理如果某个 checkpoint 已经被标记为完成但commit还没执行完比如 JobManager 在发出通知后立刻宕机恢复时会调用recoverAndCommit(txn)把这些事务补交上。这是保证不丢的关键一环。如果某个 checkpoint 没有完成状态里存在但 JobManager 那边没有 completed 记录恢复时会调用recoverAndAbort(txn)把这些事务回滚掉。这是保证不重的关键一环。判断依据就是 checkpoint id 和已完成 checkpoint 列表的比对。这个逻辑在TwoPhaseCommitSinkFunction.initializeState里实现具体做法是把状态里的事务按 checkpoint id 排序落在已完成区间内的走 commit落在未完成区间内的走 abort。这里有个实战经验如果 checkpoint 存储本身不可靠这套恢复逻辑就废了。比如你把 checkpoint 存在 JobManager 内存里JobManagerCheckpointStorageJobManager 一挂状态和事务记录全丢重启后既不知道该 commit 谁也不知道该 abort 谁外部系统里就会留下一堆永远不结束的悬挂事务。生产环境老老实实配 HDFS 或者对象存储这不是最好有是必须有。4. Kafka Sink 是最标准的参考实现也是最好的教材4.1 transactional.id 的生成规则决定了事务不会串味Kafka 的事务是靠transactional.id做隔离的。这个 ID 是生产者实例的唯一身份Kafka 用它来 fencing隔离旧的、可能还在活动的生产者实例。Flink 的 Kafka Sink 不会让你直接指定完整的transactional.id你只能配一个前缀setTransactionalIdPrefix。最终生成的 ID 是前缀 子任务索引 自增序号的组合而且这个自增序号本身也保存在状态里NextTransactionalIdHint随 checkpoint 一起持久化。为什么要这么设计因为如果每次恢复都用同一个transactional.id新的 producer 会去恢复那个 ID 下未完成的事务这本身没问题但如果transactional.id重复且旧实例还活着比如网络分区导致的脑裂Kafka 就会把其中一个 fencing 掉作业直接报ProducerFencedException。加上自增序号之后每次新事务拿到的 ID 都是全新的从根上避免了复用冲突。这里有个我踩过的坑曾经为了图省事多个作业共用同一个transactionalIdPrefix。结果两个作业的分区分配一重叠就开始互相 fencing日志里刷屏一样报错。每个启用了事务的 Sink 必须有自己的前缀前缀里最好带上作业名和版本号。4.2 事务超时是压垮线上作业的第一杀手这是我认为整个 2PC 链路里最容易出事的地方。Kafka 生产者有两个超时参数客户端侧的transaction.timeout.msbroker 侧的transaction.max.timeout.ms。客户端提交的事务超时时间如果超过 broker 侧的上限broker 会直接拒绝这个事务。而 Flink 的 Kafka 连接器默认会把客户端超时设成 1 小时Kafka broker 默认上限通常是 15 分钟——开箱即用的情况下这两者是打架的你必须先把 broker 侧的上限调大否则启用 exactly-once 的第一天就会挂。更隐蔽的是另一面如果事务超时时间设得太短比如按默认的 60 秒而你的 checkpoint 间隔是 3 分钟或者作业正在做一次耗时很长的全量恢复那么在事务提交之前broker 就会认为这个事务超时了主动把它中止掉。生产者下一次操作时收到InvalidTxnStateException或者直接进入致命错误状态整个作业挂掉且无法自动恢复。我在几个项目里总结出来的配置规则大致是这样的配置项建议值原因单次 checkpoint 间隔1 到 3 分钟太短会让事务过小、频繁提交太长会让恢复时间变长最大并发 checkpoint 数1并发 checkpoint 会拉长事务悬挂时间除非你确实需要Kafka 客户端transaction.timeout.ms≥ checkpoint 间隔 × (并发数 1) × 2给 checkpoint 和数据积压留足余量broker 端transaction.max.timeout.ms≥ 客户端设的值否则事务在提交前就被 broker 拒绝作业最大预期恢复时间必须小于事务超时恢复期间事务一直悬着超时就会被中止特别提醒作业最大预期恢复时间这一项经常被忽略。作业重启后从 checkpoint 恢复大状态可能要好几分钟这段重启期间旧事务是没有提交的。如果事务超时时间只够覆盖正常运行的 checkpoint 间隔一次意外重启就能让所有悬挂事务过期。4.3 消费端必须打开 read_committed这一条很多人会忘。Kafka 事务提交前消息对消费者是可见但不可读的。默认的isolation.level是read_uncommitted消费者会看到那些还没提交的消息一旦事务被 abort这些消息就凭空消失了或者更糟——你看到了两遍。所以下游如果也是 Flink 作业必须在 Kafka Source 上设置isolation.level read_committed。这个参数一开消费者只会读到已经提交的事务里的消息配合 Kafka 的 Last Stable Offset 机制未提交事务之后的消息也会被暂时挡住直到事务尘埃落定。顺带说一句Kafka 的事务提交消息本身也是写进 topic 的只是它是一个控制批次不会被普通消费者读到在read_committed模式下。这也是为什么 Kafka 的事务开销不算特别小高频提交事务对吞吐影响是实打实的。5. 自己写一个 2PC SinkXA 为例的完整骨架与踩点5.1 状态里能放什么、绝对不能放什么很多人在写自定义 Sink 时第一反应是把数据库连接放进事务句柄里然后在状态里存这个连接对象。这件事一定会失败。原因很简单TwoPhaseCommitSinkFunction的第二个泛型参数TXN必须是可序列化的因为pendingCommitTransactions会随 checkpoint 一起写进状态存储。而java.sql.Connection是不可序列化的KafkaProducer同样不可序列化。正确的做法是存重建事务所需的最小信息——比如 XA 事务的Xid包含 formatId、全局事务 ID、分支限定符三个字段都是基本类型恢复时用这些信息重新构造 XAResource再执行 prepare 之后的提交或回滚。public class MyXid implements Xid, Serializable { private final int formatId; private final byte[] globalTransactionId; private final byte[] branchQualifier; public MyXid(int formatId, byte[] gtrid, byte[] bqual) { this.formatId formatId; this.globalTransactionId gtrid; this.branchQualifier bqual; } Override public int getFormatId() { return formatId; } Override public byte[] getGlobalTransactionId() { return globalTransactionId; } Override public byte[] getBranchQualifier() { return branchQualifier; } }序列化器也要换成 Kryo 或者自定义的TypeSerializer因为 Flink 默认的序列化器对自定义类处理起来不够好用。老接口的构造方法大概是这个样子public class XaTwoPhaseSink extends TwoPhaseCommitSinkFunctionRow, MyTxnHandle, Void { public XaTwoPhaseSink() { super( new KryoSerializer(MyTxnHandle.class, new ExecutionConfig()), VoidSerializer.INSTANCE ); } // ... }5.2 preCommit 里到底该做哪一步这是整个实现里最需要想清楚的地方。XA 协议本身就提供了prepare阶段看上去和preCommit完美对应直接调xaResource.prepare(xid)就行了。但这里面有个陷阱。prepare之后数据库那边的分支事务处于准备好了、等待协调者指令的状态锁一直持有。如果prepare卡住很久比如连接池打满、数据库抖动整个 Sink 的写数据主线程就被阻塞了。所以实践中我更倾向于在preCommit里做轻量的收尾动作——把缓冲区 flush 到中间的临时表或者临时文件真正耗时的prepare和commit放到 commit 阶段去做并且加超时控制。Override protected void preCommit(MyTxnHandle handle) throws Exception { // 只做封口和 flush不做重量级操作 handle.flushBuffer(); } Override protected void commit(MyTxnHandle handle) { try { XAResource resource handle.openResource(); resource.prepare(handle.getXid()); resource.commit(handle.getXid(), false); handle.closeQuietly(); } catch (Exception e) { // 关键commit 失败要让 checkpoint 感知到 throw new RuntimeException(commit failed for xid handle.getXid(), e); } }注意commit里抛异常的处理。如果提交失败却不抛出作业会认为一切正常数据就真丢了。但抛出异常会导致作业重启重启后recoverAndCommit还会再试一次——所以整个 commit 路径必须幂等数据库侧的 XA 事务重复提交会返回XAER_NOTA事务不存在这个错误码要当成已经提交过来处理不能当成失败。5.3 幂等兜底为什么 commit 必须可以重复执行上面反复提到幂等这里集中说一下为什么。notifyCheckpointComplete这个通知不保证一定送达。JobManager 在发出通知的过程中宕机这个通知就丢了。听起来很吓人但实际上不影响正确性——因为那些没被通知到的事务还在状态里下次恢复时recoverAndCommit会补上。但如果通知重复送达呢同样可能发生。Flink 在某些版本里会对通知做重试加上恢复路径的重放同一个事务可能被 commit 两次。如果你的commit实现是发一条 MQ 消息给下游这种有副作用的动作重复执行就会出问题。private static final SetString COMMITTED ConcurrentHashMap.newKeySet(); Override protected void commit(MyTxnHandle handle) { String key handle.getXid().toString(); if (!COMMITTED.add(key)) { return; // 已经提交过了直接跳过 } // ... 真正提交 }用内存 Set 只是个示意生产环境里应该依赖外部系统自身的幂等能力比如数据库的唯一约束、Kafka 事务的重复提交返回XAER_NOTA、对象存储的覆盖写。6. 文档里翻不到的那几个坑从 checkpoint 存储到并发检查点6.1 Checkpoint 存储选内存等于把事务状态押给了 JobManager前面提过一次这里再强调一遍因为它的后果比想象中严重。JobManagerCheckpointStorage把 checkpoint 数据存在 JobManager 的堆内存里好处是快坏处是一旦 JobManager 进程挂掉所有已完成和未完成的 checkpoint 记录一起消失。对于 2PC 来说这意味着悬挂在外部系统里的事务既不会被提交也不会被回滚会一直占用锁和资源直到超过事务超时被强制中止作业重启后无法从任何 checkpoint 恢复只能从零开始那些悬挂事务就永远成了孤儿。生产环境必须用文件系统类存储。HDFS 是很多公司的默认选择但如果你在云上对象存储S3、OSS 这类也很常见。这里有个额外的坑对象存储的 rename 不是原子操作它是 copy delete。文件系统类 Sink 通常靠 rename 来实现两阶段提交写.inprogress文件 → commit 时转正在对象存储上这个转正过程会有短暂的两个文件都存在或者中间态可被读到的窗口。如果你的下游是直接扫目录的批处理任务建议配合_SUCCESS标记文件一起判断。6.2 并发 checkpoint 数与事务超时的乘法关系execution.checkpointing.max-concurrent-checkpoints默认是 1很多人不知道这个值可以大于 1也不知道它在 2PC 场景下的影响。设成 1 的时候任意时刻只有一个 checkpoint 在跑事务悬挂的时间大约是一个 checkpoint 的完整耗时。设成 2 之后前一个 checkpoint 还没完成后一个已经开始了事务悬挂的时间可能翻倍而且pendingCommitTransactions里会同时存在多个 checkpoint 的事务。如果这时候事务超时时间没跟着调大就会出现前一个事务被 broker 中止的故障。我的建议是除非你的 checkpoint 特别慢、必须用并发来掩盖否则保持 1。真要用并发事务超时时间至少按并发数等比放大。另外还有个相关的坑checkpoint 的对齐时间。如果某个算子的输入通道数据倾斜严重barrier 对齐可能要等很久这段时间当前事务一直处于写了但没 preCommit的状态。开启非对齐 checkpoint 能缓解这个问题因为 barrier 不再等待数据对齐但代价是 checkpoint 体积会变大在途数据也要快照。6.3 JDBC 类 Sink 为什么天然做不到端到端精确一次热词里常出现Flink 的 JDBC 连接器异常这里面相当一部分根因就跟事务语义有关。Flink 官方的 JDBC Sink 走的是批量攒批 预编译语句的路子它不是事务性的。你可以说它在某些数据库驱动下支持autoCommitfalse的批量提交但这个提交和 Flink 的 checkpoint 没有任何绑定关系——它可能在 checkpoint 完成之前就把数据提交了也可能在 checkpoint 失败后数据已经落库。那怎么办两条路第一条是幂等写入。给目标表建立唯一键SQL 改成INSERT ... ON DUPLICATE KEY UPDATEMySQL或者ON CONFLICT DO UPDATEPostgreSQL重复写的结果和写一次一样。这条路简单直接绝大多数场景够用代价是吞吐会打折因为每条记录都要走一次唯一性检查。第二条是自己在 JDBC Sink 之上实现 2PC用 XA 连接。代价是复杂度和延迟都会显著上升而且不是所有数据库的 XA 实现都足够可靠。如果你用的是 Doris、StarRocks 这类分析型数据库的连接器情况又不一样。它们一般提供自己的一套两阶段导入机制比如先写临时标签再统一生效需要看对应连接器版本文档里对 exactly-once 的支持程度。不同连接器对事务语义的支持差异非常大用了 Flink 就自动精确一次是个危险的假设接手一个新连接器时第一件事就是查它的语义保证级别。6.4 notifyCheckpointComplete 不保证送达但这不影响正确性这一点单独拿出来说是因为它经常被人误读成notifyCheckpointComplete 不可靠所以 2PC 不可靠。实际上这两件事是分开的通知只是加速提交的优化路径真正的正确性保障在状态里。只要pendingCommitTransactions被正确持久化哪怕所有通知都丢了作业下次重启时也会通过recoverAndCommit把该提交的事务补上。数据只会晚一点对外可见不会丢。真正需要担心的反而是另一面通知送达了但 commit 失败了。这种情况下事务还挂在外部系统里如果 checkpoint 已经被 subsumed被后续的 checkpoint 清理掉恢复时就找不到这个事务了。所以 commit 失败一定要抛出异常让作业重启绝不能吞掉。7. 什么时候该放弃 2PC幂等写入和 WAL 的更划算选择7.1 幂等写入的成本核算2PC 不是免费的午餐。它带来的额外开销包括每个 checkpoint 周期一次批量提交的网络往返、外部系统的锁竞争、事务日志的持久化开销以及事务超时带来的运维复杂度。如果你的 Sink 天然支持幂等按主键覆盖、按唯一约束去重、版本号覆盖那用幂等方案往往更划算。整个链路变成 at-least-once但结果等价于 exactly-once。实现上只需要在算子里维护一个可以重放的输出流写好 SQL 或者 API 调用即可。判断标准很实用问自己一句同一条数据写两次结果会不会不一样如果不会就上幂等如果会比如累加、计数、发消息通知才需要考虑 2PC 或者别的补偿机制。方案语义对下游要求延迟开销适用场景两阶段提交精确一次必须支持事务高受 checkpoint 间隔制约金融流水、订单主表、不能重复的通知幂等写入至少一次但结果等价有唯一键或覆盖写能力低维表同步、日志汇聚、指标明细WAL 中转精确一次有一个可靠的中间存储中多一次搬运目标系统不支持事务但能接受最终一致7.2 WAL 方案的适用边界WAL 的思路是Sink 不直接写目标系统先把数据写到一个高可靠的中间存储通常是 HDFS 或者对象存储上的文件等 checkpoint 成功后再由一个独立的搬运任务把文件内容导入目标系统。导入过程靠文件的命名或者目录结构区分已提交和未提交导入任务只处理已提交的文件并记录自己的处理进度。这条路的典型代表就是文件系统类 Sink。它在很多场景下比 2PC 更省心因为中间存储的能力是可控的不依赖目标系统的事务支持。代价是数据可见性有延迟至少一个 checkpoint 周期以及多维护一套搬运逻辑。我在一个日增量几十亿条的项目里最终选的就是 WAL 路线因为目标系统当时还不支持跨会话的长事务。回过头看这个选择的判断依据很简单先看目标系统能给什么再决定用哪套语义而不是反过来让业务去凑技术方案。最后分享两个我在实际运维里养成的习惯。第一个是给每个启用了事务语义的 Sink 单独建一张台账记录它的 checkpoint 间隔、事务超时、重启耗时上限任何一项改动都同步更新——这三个数是有乘法关系的改一个忘了另一个就容易出事。第二个是上线前手动做一次杀进程演练作业跑着的时候直接 kill 掉 TaskManager观察外部系统里有没有悬挂事务、重启后数据有没有重复或者丢失。这个演练花二十分钟比在生产上排查半天要划算得多。