企业官网建设流程全解析

最近把一个实时链路从"尽力而为"往"精确一次"上推，踩的坑几乎全集中在 Checkpoint 和外部系统的交界处。Flink 的 Checkpoint 本身解决的是作业内部状态的一致性——算子状态、KeyedState、OperatorState 都能靠一次全局快照对齐。但数据一旦要写进 Kafka、MySQL、Doris 这类外部系统，"内部一致"就不等于"端到端一致"了：作业重启后状态回滚了，可外部系统里那批数据已经落库，重复就发生了。这时候就得靠 Flink CheckPoint 之两阶段提交协议（Two-Phase Commit Protocol）来兜底——先在外部系统里"预占位"，等 Checkpoint 真正完成后再"转正"，中途挂掉就整批回滚。

这篇东西不打算复述官方文档，而是把我自己在生产里配过、调过、翻过车的部分完整摊开：两阶段提交在 Flink 里的实现骨架长什么样，Kafka Sink 内置的事务机制怎么和 Checkpoint 对齐，手写一个 MySQL 2PC Sink 的完整代码和参数怎么定，以及那些只会出现在日志里的报错到底在说什么。适合已经能跑通 Flink 作业、想把一致性语义从 at-least-once 提到 exactly-once 的同学；如果你刚开始学 Flink，看到"检查点间隔""事务超时"这些词有点懵，也可以先看第一、二节，把机制吃透再动手。

1. 端到端精确一次到底卡在哪一步

1.1 三种一致性语义的真实边界

先把概念理清楚，不然配参数全是瞎猜。Flink 里说的"一致性语义"其实分两层，很多人混着讲，结果调优时找不到北。

第一层是作业内部状态一致性。这一层 Flink 自己就能保证：Checkpoint 触发时，Source 记录当前消费位点，算子把自己的状态快照出去，所有快照拼成一个全局一致的 Checkpoint。失败重启时从最近一次成功的 Checkpoint 恢复，内部状态不会错乱。这一层和你用不用事务完全无关。

第二层是端到端一致性，也就是"从 Source 读进来的数据，经过处理，写进 Sink，整体上恰好处理一次"。这一层光靠 Checkpoint 是不够的，因为 Sink 写出去的数据在 Flink 状态之外，回滚 Checkpoint 不会把已经发出去的消息收回来。

于是就有了三种语义：

注意：exactly-once 说的是"Flink 这套链路内部恰好一次"。如果 Sink 是 MySQL，而你的业务代码在别处也往同一张表写，那整体还是可能重复。别把 exactly-once 当成万能承诺。

1.2 为什么单靠幂等写入不够

有人会问：既然重复会带来问题，那我用INSERT ... ON DUPLICATE KEY UPDATE做幂等写入不就行了，何必搞两阶段提交？

这个思路在很多场景下确实够用，而且成本低得多。但它有两个前提：一是数据必须有天然主键，比如订单号、设备 ID + 时间戳；二是下游必须支持原子 upsert。问题在于，很多实时场景的数据是聚合结果，比如"某商品每分钟的成交额"，它不是一条可以 upsert 的记录，而是一个累加值。这时候重放一次就等于多算一遍，幂等就失效了。

另一种常见场景是追加型数据，比如把处理后的明细写入 Kafka 供下游消费。Kafka 的消息没有主键，重复就是实打实的重复消费。

两阶段提交解决的正是这两类问题：它不依赖下游的幂等能力，而是在"这个 Checkpoint 到底算不算数"这件事上做文章——Checkpoint 成功，则这批数据全部可见；Checkpoint 失败，则这批数据全部不可见。要么全有，要么全无，这就是原子性。

代价也很明确：延迟。数据必须先"预写"，等下一次 Checkpoint 完成才能对下游可见。如果你的 Checkpoint 间隔是 1 分钟，那么最坏情况下数据要等 1 分多钟才能被下游读到。对延迟敏感的下游，这一条就足以否决整个方案。

1.3 Checkpoint 存储为什么必须是共享存储

顺带说一个被问得最多的问题：Flink 是不是一定要 HDFS？

严格说不是"一定"，但 Checkpoint 的存储必须满足两个条件：所有 TaskManager 都能访问，且作业重启后依然存在。本地文件系统只在单机伪分布式下勉强能用，一旦集群多节点部署，TaskManager 各自写本地目录，恢复时另一个节点读不到，直接报找不到 Checkpoint 元数据。

所以生产上通常选 HDFS、对象存储这类共享存储。小规模测试也可以用 NFS 挂载目录，或者高度可用的分布式文件系统。关键是别把state.checkpoints.dir配成一个只有本机能访问的路径，然后在报错里找半天。

2. 两阶段提交在 Flink 里的实现骨架

2.1 四个动作：begin、preCommit、commit、abort

Flink 的两阶段提交抽象在TwoPhaseCommitSinkFunction这个基类里（1.15 之后官方推荐迁移到 Sink V2 的Committer接口，但思路完全一致，老代码现在仍然大量存在）。它把一次完整的事务拆成四个动作，对应 Checkpoint 生命周期的四个时刻：

• beginTransaction：开启一个新事务，拿到事务句柄（比如数据库连接、Kafka 的 transactionalId）。它在算子初始化时和每次commit/abort之后被调用。
• preCommit：在 Checkpoint 触发时调用。把当前事务里攒下的数据"预提交"，同时把事务句柄写进算子状态，随 Checkpoint 一起持久化。
• commit：在notifyCheckpointComplete回调里调用，也就是 Checkpoint 被 JobManager 确认完成后。这一步才让数据真正对外可见。
• abort：Checkpoint 失败或作业取消时调用，丢弃当前事务里未提交的数据。

这四个动作的调用顺序，就是保证原子性的全部秘密。你可以把它类比成银行转账：钱先从 A 账户扣走进入"冻结中"状态（preCommit），等对方账户确认能收款了再真正解冻入账（commit），中途任何一步失败就把冻结的钱退回（abort）。

2.2 Checkpoint 与事务的时序对齐

光看方法名还是抽象，把时间轴拉出来就清楚了。假设 Checkpoint 间隔 30 秒，作业从启动到第三次 Checkpoint：

关键点在于：Checkpoint 成功之前，事务绝不能 commit。因为 Checkpoint 可能失败需要回滚，而一旦 commit 就没法收回了。反过来，Checkpoint 成功之后，事务必须尽快 commit，否则数据一直不可见，还会撞上事务超时。

还有一个容易忽略的细节：preCommit不只是"打个标记"，它必须把事务句柄写进ListState并随 Checkpoint 一起落盘。这样作业挂掉重启后，Flink 才能从 Checkpoint 里读出"当时有一个事务 T2 处于预提交状态"，从而决定是补 commit 还是 abort。这个"状态里保存未决事务"的设计，才是两阶段提交能在分布式环境下活下来的原因——它把事务的决策权交给了 Checkpoint 本身。

2.3 事务超时和 Checkpoint 间隔必须匹配

这是最容易踩的坑，没有之一。

事务超时（Kafka 里是transaction.timeout.ms，数据库里通常是连接空闲超时或锁等待超时）定义了"一个事务多久没动静就自动被判死刑"。如果事务超时小于 Checkpoint 间隔，就会出现这种尴尬局面：事务刚开启没多久就被服务端强制终止了，等 Checkpoint 完成去 commit 时，报"事务已过期"或者"找不到对应的事务"，数据直接丢失。

安全的下界怎么算：

transaction.timeout.ms > checkpoint.interval + 最大单次 Checkpoint 耗时 + 作业重启耗时余量

举个例子，Checkpoint 间隔 60 秒，大状态下单次 Checkpoint 可能耗时 40 秒，恢复一个作业大概 30 秒。那么事务超时至少要大于 130 秒，实际我会配到 5 分钟以上，也就是 300000 毫秒，留出足够余量。

为什么留这么多？因为在 Checkpoint 失败重试期间，当前事务会一直挂着不动。如果配了tolerable-failed-checkpoints = 3，那可能要连着失败三次才触发重启，这段时间事务一直处于"开着但没写入"的状态。超时值必须能覆盖这个最长悬挂时间。

提示：反过来，事务超时也不能配得过大。事务开太久，Kafka 侧会占用事务协调器资源，数据库侧会长时间持有锁甚至撑爆max_connections。几分钟到十几分钟是比较务实的区间。

3. 手写一个 Kafka 到 MySQL 的端到端精确一次链路

3.1 依赖和环境准备

光讲原理容易飘，直接上手写一遍最清楚。目标链路是：Kafka 读取订单消息，做简单聚合，写入 MySQL，要求端到端 exactly-once。

依赖上需要：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-jdbc</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.33</version> </dependency>

环境上，Checkpoint 存储要提前配好：

state.backend: rocksdb state.checkpoints.dir: hdfs:///flink/checkpoints state.savepoints.dir: hdfs:///flink/savepoints execution.checkpointing.interval: 60s execution.checkpointing.timeout: 10min execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.max-concurrent-checkpoints: 1 execution.checkpointing.tolerable-failed-checkpoints: 3

这里max-concurrent-checkpoints: 1是使用两阶段提交时的硬性建议。允许并发 Checkpoint 意味着可能有多个事务同时处于"预提交"状态，事务之间的提交顺序无法保证，下游如果对顺序敏感就会乱。我自己的做法是直接锁死为 1，牺牲一点 Checkpoint 吞吐换确定性。

3.2 MySQL 两阶段提交 Sink 的实现

MySQL 本身没有"事务预提交"这种语义，所以要用连接本身的事务来模拟：beginTransaction时setAutoCommit(false)并开启事务，preCommit时执行flush()但不 commit，commit时才真正connection.commit()。

public class MysqlTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<OrderStat, Connection, Void> { private final String jdbcUrl; private final String user; private final String password; public MysqlTwoPhaseCommitSink(String jdbcUrl, String user, String password) { super(new SimpleVersionedSerializer<Connection>() { @Override public int getVersion() { return 1; } @Override public byte[] serialize(Connection c) { // 连接对象不可序列化，只保存一个空标记 return new byte[0]; } @Override public Connection deserialize(int version, byte[] data) { return null; } }, VoidSerializer.INSTANCE); this.jdbcUrl = jdbcUrl; this.user = user; this.password = password; } @Override protected Connection beginTransaction() throws Exception { Connection conn = DriverManager.getConnection(jdbcUrl, user, password); conn.setAutoCommit(false); return conn; } @Override protected void invoke(Connection conn, OrderStat value, Context context) throws Exception { PreparedStatement ps = conn.prepareStatement( "INSERT INTO order_stat(order_id, amount, stat_time) VALUES(?,?,?)"); ps.setString(1, value.getOrderId()); ps.setBigDecimal(2, value.getAmount()); ps.setLong(3, value.getStatTime()); ps.executeUpdate(); ps.close(); } @Override protected void preCommit(Connection conn) throws Exception { // 不做任何提交动作，因为数据已经在事务里了 // 这里可以做 flush，确保网络缓冲区数据发出 conn.setAutoCommit(false); } @Override protected void commit(Connection conn) { try { conn.commit(); } catch (SQLException e) { throw new RuntimeException("commit failed", e); } finally { closeQuietly(conn); } } @Override protected void abort(Connection conn) { try { conn.rollback(); } catch (SQLException ignored) { } finally { closeQuietly(conn); } } }

这段代码有几个地方值得掰开说。

第一，Connection不能序列化。TwoPhaseCommitSinkFunction要求事务句柄能进状态，但 JDBC 连接是活对象，塞不进去。所以序列化器里只写一个空字节数组，靠 Flink 在恢复时重新建立连接。这也意味着恢复后的事务语义是"重新执行一遍未提交的数据"，而不是"续上原连接"。这一点必须接受，否则整个模型不成立。

第二，preCommit里几乎什么都不用做。因为 MySQL 的事务本身就有"未提交不可见"的特性，天然满足第一阶段的要求。这和 Kafka 不一样，Kafka 需要显式调flush()把缓冲消息发出去，让 broker 端持有但不标记为已提交。

第三，commit抛异常会导致作业失败并重启。这是有意的：commit 阶段失败不能静默吞掉，否则数据就丢了。让作业失败、从上一个 Checkpoint 恢复、重新走一遍流程，才是正确姿势。

3.3 主程序与参数配置

把 Sink 接进作业：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000); env.getCheckpointConfig().setCheckpointTimeout(600_000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("kafka:9092") .setTopics("order-topic") .setGroupId("order-stat-group") .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<OrderStat> stats = env .fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source") .map(new ParseAndAggregateFunction()) .name("aggregate") .uid("aggregate-uid"); stats.addSink(new MysqlTwoPhaseCommitSink( "jdbc:mysql://mysql:3306/dw?useSSL=false", "flink", "flink_pwd")) .name("mysql-2pc-sink") .uid("mysql-2pc-sink-uid"); env.execute("order-stat-exactly-once");

两个参数特别想强调。

minPauseBetweenCheckpoints设成 Checkpoint 间隔的一半，是为了给 commit 操作留出窗口期。两阶段提交的 commit 是发生在 Checkpoint 完成回调里的，如果 Checkpoint 一个接一个连轴转，commit 还没执行完下一个 Checkpoint 就来了，事务会堆积。

uid必须显式指定，而且上线后绝对不能改。Flink 靠 uid 把算子和状态做映射，uid 变了就相当于换了个算子，之前保存的事务句柄状态全部丢失。后果是：那些处于"预提交"状态的事务既不会被 commit 也不会被 abort，永久悬挂在数据库里占着锁。

3.4 怎么验证真的做到了精确一次

写完不算完，得能证明。我的验证方法是"故意制造故障 + 对数"。

第一步，正常跑 5 分钟，记录 MySQL 里的总行数，记为 A。

第二步，重启作业，让它重新消费一部分已经处理过的数据（把 Kafka 消费位点人为往前调一点），继续跑 5 分钟。

第三步，等作业稳定后再次统计行数，记为 B。如果 B 和 A 的差值恰好等于新增数据量，说明没有重复；如果 B 明显偏大，说明有两阶段提交没生效的地方；如果 B 偏小，那就更严重了，是丢数。

另一个更直接的办法是打开网络抓包或者在 Sink 的commit/abort里打点，统计两个方法被调用的次数。正常情况下，commit次数应该等于成功的 Checkpoint 次数（可能少一次，因为最后一次 Checkpoint 未必完成），abort次数应该只在故障时出现。

注意：做这类验证一定要在独立的测试环境做。调整消费位点会污染生产数据，代价可能是几小时的对账工作。

4. Kafka Sink 内置的两阶段提交与实战陷阱

4.1 Kafka 自己的事务机制怎么和 Flink 接上

Kafka 从 0.11 版本开始支持事务，Flink 的 Kafka Sink 直接复用了它。开启方式很直接：

KafkaSink<String> sink = KafkaSink.<String>builder() .setBootstrapServers("kafka:9092") .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("result-topic") .setValueSerializationSchema(new SimpleStringSchema()) .build()) .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) .setTransactionalIdPrefix("order-stat-") .setProperty("transaction.timeout.ms", "300000") .build();

底层发生了什么：每个并行 Sink 子任务会用transactionalIdPrefix + subtaskIndex拼出一个唯一的 transactionalId，向事务协调器注册。Checkpoint 触发时调producer.flush()，Agent 把缓冲的消息发给 broker，但消息处于"未提交"状态，消费者用read_committed隔离级别看不到。notifyCheckpointComplete到达后调producer.commitTransaction()，这批消息才对外可见。

transactionalIdPrefix有两个约束：同一时刻不能有两条作业用同一个前缀，否则第二个作业启动时会因为 transactionalId 冲突而失败；上线后不能随便改，改了等于换了事务身份，之前未提交的事务就成孤儿了。

4.2 那些让人抓狂的报错

报错一：InvalidProducerEpochException或者ProducerFencedException

这个报错的含义是"你的事务身份被别人抢了"。最常见的原因是同一个作业被重复提交了两次，两个实例用同样的 transactionalId 在跑。排查方向：检查调度平台上是否有残留的僵尸作业，或者作业重启时旧实例还没完全退出。

报错二：InvalidTxnStateException或者commitTransaction超时

一般是事务超时了。要么是transaction.timeout.ms配得太小，要么是 Checkpoint 长时间卡住导致事务悬挂超过阈值。我遇到过一次正是下游 HDFS 集群抖动，Checkpoint 卡了 8 分钟，直接超了默认的 5 分钟事务超时。解决办法有两个：调大超时值，或者缩短 Checkpoint 超时时间让它快速失败重试。

报错三：下游一直读不到数据

事务提交了但下游看不到，八成是消费者用的隔离级别不对。默认的read_uncommitted或不设置隔离级别时行为不确定，正确做法是显式配置：

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

用read_committed之后，消费者只能看到已提交事务的消息，未提交的会被过滤掉。代价是会有一定的读取延迟。

报错四：TimeoutException: Expiring N record(s)

消息在缓冲区里等太久被丢弃了。检查linger.ms和batch.size的配合，也检查事务里是否积累了过多数据导致发送耗时超过阈值。两阶段提交下，一个事务里塞的数据量不宜过大——事务跨越时间越长，占用的缓冲区、锁和协调器资源越多。

5. 常见问题与排查技巧实录

5.1 问题速查表

5.2 几条只有踩过才知道的经验

第一，事务超时宁可配大不配小。配小了丢数据的代价是几小时的对账，配大了顶多多占点资源。这条经验背后是一次真实的教训：我们把事务超时配成了 2 分钟，Checkpoint 间隔 1 分钟，理论上是够的，但一次机房网络抖动让单次 Checkpoint 花了 3 分半，结果那一批数据在 commit 时直接报事务不存在。损失不大，但定位花了整整一个下午。

第二，Sink 的并行度不要盲目调大。每个并行子任务对应一个独立事务、一个独立连接。并行度调到 16，就意味着最多有 16 个事务同时挂在数据库上，连接池配置跟不上就是一堆Too many connections。除非下游确实扛不住写入压力，否则 Sink 并行度保持在 2 到 4 是比较稳的。

第三，abort 一定要做资源兜底。很多人写abort时只调了rollback()，忘了关连接。作业频繁重启时，这些没关掉的连接会累积，最终把数据库连接数吃满。正确写法是try { rollback } catch {} finally { close }，关连接这一步不能省。

第四，不要在 Sink 里做重业务逻辑。我见过有人在invoke里调用外部 HTTP 接口做数据补全，一旦接口超时，整个事务就卡住了，Checkpoint 跟着失败。两阶段提交的窗口期很宝贵，invoke里只应该做纯粹的写入动作，任何可能阻塞的操作都要挪到上游算子。

第五，测试环境一定要模拟故障。生产上第一次遇到 commit 阶段崩溃时，如果没演练过，心态很容易崩。我的做法是在测试环境用kill -9直接杀 TaskManager，观察作业重启后数据库里的事务是被正确提交还是回滚，是否有残留。跑通几次之后，对这套机制的行为就心里有数了。

6. 选型：什么时候该上两阶段提交

6.1 幂等写入和两阶段提交的取舍

不是所有场景都值得上两阶段提交。判断标准其实很简单，问自己两个问题：数据有没有天然主键？下游能不能接受 1 到 2 个 Checkpoint 间隔的可见延迟？

如果数据有天然主键、下游支持 upsert，那用幂等写入就够了，简单、延迟低、还不用管事务超时这一堆麻烦。比如订单表的同步，订单号本身就是主键，直接INSERT ... ON DUPLICATE KEY UPDATE完事。

如果数据是聚合值、是追加型消息、或者下游明确要求不能看到中间态，那就得上两阶段提交。指标类、对账类、计费类业务基本都属于这一类。

还有一类是"混合方案"，值得单独提一句：Source 端保证精确一次，Sink 端用幂等。Kafka Source 本身能通过位点提交保证精确一次，Sink 端如果下游有主键就做幂等。这种组合在很多场景下已经够了，比全链路 2PC 简单得多。

6.2 各类连接器的现状与坑点

实际选型时，最大的痛点是不同连接器对两阶段提交的支持程度差异极大，而且这个差异在文档里往往一句话带过。

Kafka 连接器支持最好，内置DeliveryGuarantee.EXACTLY_ONCE，开箱可用，也是我推荐的入门练习对象。

JDBC 连接器的情况要复杂一些。官方 JDBC Sink 在较早版本里对 exactly-once 的支持是通过TwoPhaseCommitSinkFunction实现的，但需要数据库开启 XA 支持（比如 MySQL 的 XA 事务）。而 XA 事务在生产里有很多争议：性能损耗明显、长时间悬挂的 XA 事务需要 DBA 手动清理、某些云数据库甚至不开放 XA 权限。我自己的经验是，除非业务强需求，否则 JDBC 链路优先用幂等 + 主键的方案，而不是硬上 2PC。如果用 JDBC 遇到了连接层面的异常，比如驱动版本与数据库版本不匹配导致的元数据读取错误，先确认驱动版本，再确认连接串参数，最后才怀疑两阶段提交的配置。

Doris 连接器这几年的成熟度提升很快。它的 Sink 采用 Stream Load 方式写入，本身通过 Label 机制做幂等——同一个 Label 的导入请求重复提交会被 Doris 自动去重。所以它的 exactly-once 实现思路和两阶段提交不完全一样，更偏向"用 Label 做幂等"。需要注意的是，用 Flink CDC 写入 Doris 时经常遇到类型映射问题，比如源端是日期类型，Doris 侧字段类型不匹配，报出类似"类型不一致"的元数据错误。这类问题本质上是类型系统对不上，和事务机制无关，但排查时容易和一致性配置混在一起，建议先单独把类型对齐，再验证一致性。

TiDB 通过 Flink SQL 写入的链路，因为有分布式事务支持，理论上是比较容易做精确一次的。用 Flink SQL 的话，sink表配置里开启相关语义就行，但要注意 TiDB 侧事务大小限制，一批写入量过大时会报事务过大失败。

提示：选连接器之前，先去对应版本的官方文档确认它到底支持哪种语义。很多"数据重复"的锅最后都不是两阶段提交的配置问题，而是连接器压根就没提供 exactly-once 能力。

6.3 一个务实的落地顺序

如果现在就要把这个方案推上线，我建议按这个顺序来，别一上来就全链路开 2PC。

先做一轮链路梳理，把所有写入口列出来，确认哪些是主链路、哪些是旁路。然后从主链路里挑一条数据量适中的，先开两阶段提交，用小流量跑一周，观察 Checkpoint 成功率、commit 耗时、下游可见延迟这三个指标。指标稳定了再逐步扩大范围。

同时把监控补上。要盯的东西包括：Checkpoint 成功率与平均耗时、Sink 的 commit 与 abort 调用次数（可选）、外部系统里的未决事务数量、数据库连接数。这几个指标里，未决事务数量是最灵敏的预警信号，一旦持续增长就说明 commit 环节出了问题，得赶紧查。

最后再分享一个小技巧。两阶段提交最难排查的情况是"事务悬挂"，因为它不报错、不告警，只是数据凭空少了。我的做法是在 Sink 里给每个事务起一个可读的标识（比如jobName-checkpointId-subtaskIndex），并把这个标识写进外部系统的备注字段或者日志。出问题时，直接拿这个标识去外部系统里搜，能立刻定位到是哪个 Checkpoint、哪个子任务的事务没被处理，比翻 Flink 日志快得多。这个标识的生成逻辑要放在beginTransaction里，并且随 Checkpoint 一起持久化，保证重启后依然能对上。