3. 两阶段提交协议：Flink 如何用 Two-Phase-Commit 实现 Sink 端的 Exactly-Once

聊到 Flink 的 Exactly-Once 语义，很多人第一反应就是 Checkpoint。没错，Checkpoint 是 Flink 的「命根子」，它保证了算子状态的精确一次恢复。但问题来了——状态恢复之后，数据怎么写到外部系统？

你想想看，Flink 内部可以靠 Checkpoint 把状态回滚到某个一致点，可 Kafka 已经收到了之前的数据，它不会自己「撤回」。这就尴尬了：Flink 重启后重新发一遍数据，Kafka 里就多了一份重复消息。

所以，Sink 端的 Exactly-Once 才是真正的硬骨头。Flink 解决这个问题的方案，就是 Two-Phase-Commit（两阶段提交）协议。说白了，就是让 Flink 和外部系统「商量着来」——先预提交，确认大家都准备好了，再正式提交。

核心观点：Two-Phase-Commit 是 Flink 实现端到端 Exactly-Once 的「最后一公里」。没有它，Checkpoint 再完美，数据到了外部系统还是会重复。

3.1 两阶段提交协议的本质

两阶段提交，名字听着挺唬人，其实逻辑很简单。它把一个分布式事务拆成两步：

  1. 预提交阶段（Pre-Commit）：协调者问所有参与者：「我要提交了，你们准备好了吗？」参与者各自检查，能提交就回复「OK」，不能就回复「NO」。
  2. 提交阶段（Commit）：如果所有人都说「OK」，协调者就正式通知大家提交。如果有人说了「NO」，协调者就通知大家回滚。

Flink 的 Two-Phase-Commit 实现，本质上就是这个思路。只不过，Flink 把 Checkpoint 的 Barrier 当成了「协调者」的指令，把 Sink 算子当成了「参与者」。

我在项目中第一次接触这个机制时，总觉得它太「重」了。后来踩了坑才明白——在分布式系统里，想做到 Exactly-Once，没有「轻量级」的捷径。两阶段提交虽然开销大，但它是最可靠的方案。

3.2 Flink 的 Two-Phase-Commit 抽象：TwoPhaseCommitSinkFunction

Flink 提供了一个通用的抽象类 TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>，专门用来实现 Sink 端的 Exactly-Once。你只需要实现几个关键方法，Flink 就会自动帮你完成两阶段提交的流程。

这个类的三个泛型参数分别是：

参数 含义 举例（Kafka Sink）
IN 输入数据类型 String
TXN 事务句柄类型 FlinkKafkaProducer.TransactionState
CONTEXT 上下文类型 FlinkKafkaProducer.Context

你需要重写的方法有这几个：

  • beginTransaction()：开启一个新事务。返回一个事务句柄。
  • invoke(txn, value)：将数据写入当前事务。注意，这里只是「写入事务缓冲区」，数据还没真正提交到外部系统。
  • preCommit(txn)：预提交。Flink 会在 Checkpoint 触发时调用这个方法。你在这里做「预提交」操作，比如让 Kafka Producer 开始提交事务。
  • commit(txn)：正式提交。Flink 确认 Checkpoint 成功后，会调用这个方法。你在这里做「正式提交」，比如让 Kafka Producer 完成事务提交。
  • abort(txn)：回滚。如果 Checkpoint 失败或作业失败，Flink 会调用这个方法。你在这里做「回滚」，比如让 Kafka Producer 中止事务。

我的经验：刚开始写自定义 Sink 时，我总把 invoke 里的写入操作直接提交到外部系统。结果 Checkpoint 恢复后数据全重复了。记住：invoke 只是「写缓冲区」，真正的提交在 commit 里。

3.3 以 Kafka Sink 为例：完整的两阶段提交流程

Kafka 从 0.11 版本开始支持事务（Transactional Writes），这正好给 Flink 的两阶段提交提供了「基础设施」。Flink 的 FlinkKafkaProducer（注意是旧版 API，新版用 KafkaSink）就是基于 TwoPhaseCommitSinkFunction 实现的。

我们来看一个简化版的流程：

// 伪代码：FlinkKafkaProducer 的两阶段提交实现
public class FlinkKafkaProducer<T> extends TwoPhaseCommitSinkFunction<T, KafkaTransactionState, KafkaTransactionContext> {

    @Override
    protected KafkaTransactionState beginTransaction() throws Exception {
        // 1. 创建一个新的 Kafka Producer 实例
        // 2. 初始化一个事务（producer.initTransactions()）
        // 3. 返回事务状态对象
        return new KafkaTransactionState(producer, transactionalId);
    }

    @Override
    public void invoke(KafkaTransactionState transaction, T value, Context context) throws Exception {
        // 将数据写入 Kafka 事务缓冲区
        // 注意：此时数据不会对外可见
        transaction.producer.send(new ProducerRecord<>(topic, value));
    }

    @Override
    protected void preCommit(KafkaTransactionState transaction) throws Exception {
        // 预提交：flush 缓冲区，但不要 commit
        // Kafka 的 preCommit 其实就是 producer.flush()
        transaction.producer.flush();
        // 注意：这里不能调用 producer.commitTransaction()
    }

    @Override
    protected void commit(KafkaTransactionState transaction) throws Exception {
        // 正式提交：提交 Kafka 事务
        // 调用 producer.commitTransaction() 后，数据才对消费者可见
        transaction.producer.commitTransaction();
    }

    @Override
    protected void abort(KafkaTransactionState transaction) throws Exception {
        // 回滚：中止 Kafka 事务
        transaction.producer.abortTransaction();
    }
}

这个流程配合 Checkpoint 的执行时机，就构成了完整的 Exactly-Once 链路。我画了一张图，帮你理清整个时序：

Flink Two-Phase-Commit + Kafka Sink 时序图 Flink 算子 Checkpoint Sink (Txn) Kafka beginTransaction() Barrier 到达 invoke(数据) 发送数据 事务缓冲区 preCommit() Checkpoint 触发 flush() Checkpoint 完成 commit() commitTransaction() 数据对外可见 abort() Checkpoint 失败 abortTransaction() 说明：虚线表示 Checkpoint 失败时的回滚路径。abort() 会清理事务缓冲区，保证数据不残留。

这张图展示了完整的时序。注意看几个关键点：

  • Barrier 到达时，Flink 会调用 beginTransaction() 开启一个新事务。这个新事务用来接收 Barrier 之后的数据。
  • Checkpoint 触发时，Flink 调用 preCommit()。此时数据已经写入 Kafka 的事务缓冲区，但消费者看不到。
  • Checkpoint 成功后，Flink 调用 commit()。Kafka 事务正式提交，数据对外可见。
  • Checkpoint 失败时，Flink 调用 abort()。Kafka 事务被中止，数据就像没发生过一样。

我曾经踩过的坑：有一次我忘了在 preCommit 里做 flush，结果 Checkpoint 成功了，但数据还在缓冲区没发出去。commit 的时候 Kafka 发现事务里没数据，直接报错。记住：preCommit 一定要 flush，确保所有数据都进入了事务缓冲区。

3.4 为什么 Kafka 能支持 Two-Phase-Commit？

Kafka 能成为 Flink 两阶段提交的「最佳拍档」，是因为它从 0.11 版本开始引入了事务 API。这个 API 提供了三个核心能力：

Kafka 事务 API 对应 Flink 方法 作用
initTransactions() beginTransaction() 初始化事务，生成唯一的事务 ID
beginTransaction() beginTransaction() 开始一个新的事务
sendOffsetsToTransaction() （内部使用） 将消费偏移量写入事务，实现「消费-处理-写入」的原子性
commitTransaction() commit() 提交事务，数据对外可见
abortTransaction() abort() 中止事务，数据被丢弃

Kafka 的事务机制，说白了就是给 Producer 提供了一个「要么全写，要么全不写」的保证。Flink 利用这个特性，把 Checkpoint 的边界和 Kafka 事务的边界对齐了。

这里有个细节要注意：Kafka 的事务 ID 必须是唯一的。Flink 在创建 Producer 时，会为每个并行子任务生成一个唯一的事务 ID，格式通常是 ${taskName}-${subtaskIndex}-${restartCount}。这样即使作业重启，也不会出现事务 ID 冲突。

3.5 两阶段提交的局限性

说了这么多优点，我也得泼点冷水。两阶段提交不是万能的，它有几个硬伤：

  • 外部系统必须支持事务：不是所有系统都像 Kafka 一样有事务 API。比如 Elasticsearch、HBase 的旧版本就不支持。这时候你只能用「幂等写入」或「去重」来近似实现 Exactly-Once。
  • 性能开销大：每个 Checkpoint 都要做一次两阶段提交，事务的开启和提交都有额外开销。如果你的 Checkpoint 间隔很短（比如 1 秒），Kafka 的事务负载会很高。
  • 事务超时问题：Kafka 的事务有超时时间（默认 15 分钟）。如果 Checkpoint 卡住了，事务超时后会被 Kafka 自动中止。恢复时 Flink 会尝试提交一个已经中止的事务，导致报错。
  • 状态膨胀：Flink 需要保存未提交的事务句柄。如果 Checkpoint 间隔长、数据量大，事务句柄会占用不少内存。

我的建议：如果你的业务场景对延迟不敏感，可以把 Checkpoint 间隔设大一点（比如 30 秒到 1 分钟），减少两阶段提交的频率。另外，Kafka 的 transaction.timeout.ms 要设得比 Checkpoint 超时时间长，避免事务被意外中止。

3.6 总结：两阶段提交的核心价值

两阶段提交协议，是 Flink 实现端到端 Exactly-Once 的「最后一块拼图」。它把 Checkpoint 的「状态一致性」扩展到了外部系统，让数据在 Flink 内部和外部系统之间保持精确一次。

Kafka Sink 是最经典的例子，但不是唯一的。只要外部系统支持事务，你都可以用 TwoPhaseCommitSinkFunction 来实现。比如 JDBC Sink、Pravega Sink，原理都是一样的。

嗯，这里要注意：两阶段提交不是银弹。如果你的外部系统不支持事务，或者你的业务场景对性能要求极高，可以考虑「幂等写入」或「去重」方案。这些我们后面会详细讲。

← 上一章 📖 返回目录 下一章 →