从“记账本”到“原子提交”：拆解 Flink 状态管理与 Exactly-Once 的底层账本逻辑

引言

先说一个我亲身踩过的坑。

2021 年我负责一个实时计费系统，用 Flink 消费 Kafka 里的用户流量日志，按分钟聚合后写入 MySQL 做扣费。上线第一周，财务对账发现每天多扣了约 0.3% 的用户费用。排查三天，定位到问题：Job 因背压触发了 TaskManager 重启，Kafka Source 从上次提交的 offset 重新消费，而 MySQL 那边上一批聚合结果已经写进去了——重复计算、重复扣费。

这就是典型的“至少一次”语义在业务侧的代价。要解决它，你必须同时搞定两件事：状态怎么持久化，以及外部系统怎么参与同一个事务。前者是 Flink 状态管理的核心，后者是 TwoPhaseCommitSinkFunction 存在的意义。这两件事咬合在一起，才构成 Exactly-Once。

这篇文章不打算复述官方文档里“什么是 KeyedState”这种内容。我想带你从 StreamTask 的 checkpoint 触发链路一路读下去，看清楚 Flink 到底是怎么把“分布式快照”这件事做得既轻量又正确的，以及为什么很多人配了 EXACTLY_ONCE 却依然重复写数据。

核心概念：先建立两个生活化直觉

类比一：状态 = 后厨的“半成品台账”

把 Flink 算子想象成餐厅后厨的一个工位。它处理订单（事件），过程中会产生半成品：切了一半的菜、正在腌的肉（这就是状态）。如果突然停电（故障），你要能恢复营业，就必须有一份台账记录“3 号单的菜切到第几刀了”。Checkpoint 就是定期给整个后厨拍一张全景照片，而不是让每个厨师自己记流水账——后者在分布式环境下根本对不齐。

类比二：Exactly-Once = 银行转账的“原子提交”

“恰好一次”不是数学意义上的“事件只被处理一次”（这在分布式系统里做不到），而是“处理效果只体现在结果里一次”。就像转账：中间可能重试了三次扣款请求，但最终账上只少了一笔钱。实现手段是：要么全部生效，要么全部回滚，且重复提交要有幂等或去重保护。

技术定义

  • State：算子在其生命周期内需要记住的数据，分为 Keyed State（按 key 分区，如 ValueState、ListState、MapState）和 Operator State（算子级别，如 Kafka Source 的 offset）。
  • Checkpoint：基于 Chandy-Lamport 算法改造的异步分布式快照，把“某一时刻全拓扑的状态”持久化到可靠存储。
  • Exactly-Once：端到端语义，= Source 可重放 + Flink 内部状态快照一致 + Sink 幂等/事务提交。三者缺一不可。

源码/原理深度分析

1. Checkpoint 的触发链路：从 JobManager 到算子

很多人以为 checkpoint 是“暂停计算、拍照、继续”。实际 Flink 是异步 barrier 对齐，几乎不停机。核心链路如下：

graph TD A[CheckpointCoordinator
定时触发] --> B[向 Source 注入 Barrier] B --> C[Source 把 offset 快照存入 StateBackend] C --> D[Barrier 随数据流向下游传播] D --> E{算子收到 Barrier} E -->|所有输入都到齐| F[异步快照本地状态
上传持久化存储] E -->|部分输入未到| G[阻塞该通道
等待对齐] F --> H[向 JobManager 上报 ACK] H --> I{全部算子 ACK?} I -->|是| J[Checkpoint 完成
通知 Sink 提交事务] I -->|否| K[超时则丢弃本次]

关键点在 CheckpointBarrierHandler。以 AlignedBarrierHandlerState 为例，当某个 input channel 先到达 barrier，它会阻塞该 channel 的后续数据，等其余 channel 的 barrier 也到齐，才放行。这就是“对齐”的代价——反压敏感。Flink 1.11 之后引入的 Unaligned Checkpoint 正是为了绕开这个阻塞：把 barrier 之前的数据也一并写进快照，代价是快照体积变大。这是一个典型的“空间换阻塞时间”的权衡。

2. StateBackend：状态到底存在哪

StateBackend 决定了两件事：状态存内存还是磁盘，以及快照怎么写。主流是 HashMapStateBackend（状态在 JVM 堆上）和 EmbeddedRocksDBStateBackend（状态在本地 RocksDB，支持超大规模）。

以 RocksDB 为例，它的 checkpoint 机制非常巧妙：利用 RocksDB 的 Checkpoint API 创建 SST 文件的硬链接，而非全量拷贝，然后把增量文件上传到 DFS。这解释了为什么 RocksDB 后端能做增量 checkpoint，而 Heap 后端只能全量。

3. 端到端 Exactly-Once 的命门：TwoPhaseCommitSinkFunction

这是最容易被误解的部分。Flink 内部状态一致了，但外部数据库不会自动“回滚”。TwoPhaseCommitSinkFunction 的设计对应了分布式事务的两阶段提交：

sequenceDiagram participant CP as CheckpointCoordinator participant Sink as Sink算子 participant EXT as 外部系统(如Kafka事务) CP->>Sink: 触发 checkpoint (pre-commit阶段) Sink->>EXT: 开启事务 / 写入预提交数据 Sink->>Sink: 把事务句柄存入 OperatorState Sink-->>CP: 上报 ACK Note over CP: 等所有算子 ACK CP->>Sink: notifyCheckpointComplete (commit阶段) Sink->>EXT: 提交事务，数据对外可见

注意两个致命细节：

  1. invoke 阶段写的数据在下游是“不可见”的（事务未提交）。所以 Sink 必须支持事务或幂等写入，否则这层封装毫无意义。
  2. notifyCheckpointComplete 可能丢失。Flink 不保证它一定被调用。因此 Sink 必须能在重启后从 OperatorState 恢复未提交的事务并重新提交——这就是为什么 TwoPhaseCommitSinkFunction 要求实现 recoverAndCommit。

4. 为什么配了 EXACTLY_ONCE 还会重复？

因为 CheckpointingMode.EXACTLY_ONCE 只保证 Flink 内部的状态一致。如果你的 Sink 是普通 JDBC 写入，它不在事务保护范围内，重启后照样重复。这个坑我见过太多次：配置项名字太有迷惑性。

实战代码

示例 1：Keyed State 的正确打开方式（含 TTL 与状态清理）

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 场景：统计每个用户最近 1 小时内的访问次数，
 * 并要求状态在 2 小时无更新后自动清理，防止状态无限膨胀。
 */
public class UserVisitCounter extends KeyedProcessFunction<String, Long, String> {

    // 声明状态句柄（注意：只声明，不 new）
    private transient ValueState<Long> visitCount;
    private transient ValueState<Long> lastVisitTs;

    @Override
    public void open(Configuration parameters) {
        // 状态描述符：名字 + 类型 + TTL 策略
        ValueStateDescriptor<Long> countDesc =
                new ValueStateDescriptor<>("visitCount", Long.class);
        // TTL 是生产环境的救命稻草：2 小时无访问即清理
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.hours(2))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .cleanupInRocksdbCompactFilter(1000) // RocksDB 压缩时过滤，降低开销
                .build();
        countDesc.enableTimeToLive(ttlConfig);
        visitCount = getRuntimeContext().getState(countDesc);

        lastVisitTs = getRuntimeContext().getState(
                new ValueStateDescriptor<>("lastVisitTs", Long.class));
    }

    @Override
    public void processElement(Long ts, Context ctx, Collector<String> out) throws Exception {
        Long cur = visitCount.value();
        cur = (cur == null) ? 1L : cur + 1;
        visitCount.update(cur);
        lastVisitTs.update(ts);
        // 只在 key 维度输出，下游可继续聚合
        out.collect("user=" + ctx.getCurrentKey() + ", count=" + cur);
    }
}

注释要点：状态句柄必须在 open 里通过 getRuntimeContext().getState() 获取，不能 new；TTL 的 cleanupInRocksdbCompactFilter 只对 RocksDB 后端生效，Heap 后端靠全量扫描，慎用。

示例 2：RocksDB + 增量 Checkpoint 的完整配置

import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.CheckpointingMode;

public class ExactlyOnceJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1. 开启 checkpoint：间隔 60s，恰好一次语义，容忍 1 次失败
        env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000); // 两次间隔下限
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60_000);     // 超时丢弃
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);        // 串行，避免状态抖动
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(1);

        // 2. 使用 RocksDB 后端并开启增量 checkpoint（仅 RocksDB 支持）
        EmbeddedRocksDBStateBackend rocksDB =
                new EmbeddedRocksDBStateBackend(true); // true = 启用增量
        env.setStateBackend(rocksDB);
        // 快照落到 HDFS/S3 等可靠存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints");

        // 3. 重启策略：固定延迟重试，配合 checkpoint 恢复
        env.setRestartStrategy(
                org.apache.flink.api.common.restartstrategy.RestartStrategies
                        .fixedDelayRestart(3, org.apache.flink.api.common.time.Time.seconds(10)));

        // ... 后续 source / sink 装配
        env.execute("exactly-once-demo");
    }
}

关键点：setMinPauseBetweenCheckpoints 和 setMaxConcurrentCheckpoints(1) 是防止大状态作业因 checkpoint 过于频繁拖垮吞吐的标配。增量 checkpoint 只有 RocksDB 后端支持，且首次仍为全量。

示例 3：Kafka 端到端 Exactly-Once（事务 Sink）

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

public class KafkaEOSDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30_000, org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE);

        // Source：offset 由 checkpoint 管理，重启从快照恢复
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("broker:9092")
                .setTopics("input-topic")
                .setGroupId("eos-group")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(
                        org.apache.kafka.clients.consumer.OffsetResetStrategy.EARLIEST))
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> stream = env.fromSource(
                source, WatermarkStrategy.noWatermarks(), "kafka-source");

        // Sink：开启事务，DeliveryGuarantee.EXACTLY_ONCE 才会走两阶段提交
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers("broker:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                                .setTopic("output-topic")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build())
                .setDeliveryGuarantee(
                        org.apache.flink.connector.base.DeliveryGuarantee.EXACTLY_ONCE)
                // 事务超时必须大于 checkpoint 间隔，否则事务提前过期导致数据丢失
                .setProperty("transaction.timeout.ms", "900000")
                .setTransactionalIdPrefix("flink-eos-")
                .build();

        stream.sinkTo(sink);
        env.execute("kafka-eos-demo");
    }
}

注释要点：transaction.timeout.ms 必须大于 checkpoint 间隔（建议 3 倍以上），否则 Kafka broker 会在 checkpoint 完成前把事务判为超时，导致数据丢失。这是生产事故的高发点。

方案对比：Exactly-Once 的三条技术路线

方案 实现机制 优点 缺点 适用场景
两阶段提交（2PC） Sink 开启事务，checkpoint 完成时提交 真正的端到端精确一次 依赖外部系统支持事务；事务超时难调 Kafka、支持 XA 的 DB
幂等写入 写入带唯一键，重复写覆盖 实现简单，无事务开销 依赖主键/去重键；无法处理部分更新 支持 upsert 的 OLAP（如 Doris、HBase）
去重表 / 版本号 下游记录已处理批次号 通用性强 需额外存储；有状态膨胀 自定义 Sink、异构系统

对比 Spark Structured Streaming：Spark 的 Exactly-Once 主要依赖微批 + offset 管理 + 幂等 Sink，事务支持不如 Flink 的原生 2PC 细粒度；Flink 的逐条 barrier 快照在低延迟场景更有优势，但 Unaligned Checkpoint 之前对反压敏感是它的软肋。

最佳实践与避坑指南

  1. 别迷信配置名：CheckpointingMode.EXACTLY_ONCE 只管内部。Sink 不支持事务/幂等，端到端依然是 At-Least-Once。
  2. 事务超时 > checkpoint 间隔 × 3：Kafka 事务超时是硬约束，宁可设大。
  3. 大状态作业必用 Unaligned Checkpoint：反压场景下对齐会卡死 checkpoint，导致频繁超时。
  4. TTL 是状态治理的第一步：Keyed State 不清理就是内存泄漏，用 StateTtlConfig 兜底。
  5. checkpoint 存储用高可用 DFS：本地文件系统在 JobManager 故障时直接丢失恢复能力。
  6. 监控 checkpoint 时长和大小：lastCheckpointDuration 持续上涨，说明状态在膨胀或对齐在阻塞，要提前介入。
  7. Sink 实现 recoverAndCommit：notifyCheckpointComplete 不保证送达，重启后必须能找回未提交事务。

总结

回到开头的计费系统。我们最终的修复方案是：把 MySQL 写入改成“批次号 + 唯一索引”的幂等写入，同时用 Keyed State 记录每个用户已结算的最大批次。这样即使重放，重复批次被唯一索引挡掉，效果上就是 Exactly-Once。

这件事让我对 Flink 的 Exactly-Once 有了更朴素的理解：它不是魔法，而是一份“可恢复的账本 + 可重放的输入 + 幂等/事务的输出”三方合约。Checkpoint 负责账本，Kafka offset 负责重放，Sink 的 2PC 或幂等负责收尾。任何一环偷懒，语义就会退化。

延伸思考：随着 Flink 在流批一体和 Paimon/Iceberg 等湖仓上的演进，Exactly-Once 的边界正在从“算子状态”扩展到“数据湖表格式的原子提交”。但底层逻辑没变——依然是快照与提交的舞蹈。理解了这一层，你就不会再被任何一个“EXACTLY_ONCE”配置项骗到。