从零开始学Flink：状态管理与容错机制

流式计算任务通常需要 7x24 小时长期运行，面对网络抖动、机器故障或代码 Bug，如何保证任务不挂？或者挂了之后能自动恢复且数据不丢、不重？这正是 Flink 引以为傲的资本：强大的状态管理与基于 Checkpoint 的容错机制。

本文将带你深入理解 Flink 是如何“记忆”数据的，以及它是如何在故障发生时“时光倒流”恢复现场的。

一、什么是状态（State）

在流计算中，数据是一条条流过的。如果处理一条数据时，需要依赖之前的数据（例如：计算过去一小时的总和、去重、模式匹配），那么这些“之前的数据”或“中间计算结果”就是状态。

1. 状态的分类

Flink 的状态分为两大类：Managed State（托管状态） 和 Raw State（原生状态）。我们日常开发 99% 使用的是托管状态，由 Flink 运行时自动管理内存、序列化和故障恢复。

Managed State 又细分为：

• Keyed State（键控状态）
  • 只能在 KeyedStream（即 keyBy 之后）上使用。
  • 状态是跟 Key 绑定的。Flink 为每个 Key 维护一份独立的状态实例。
  • 常用类型：ValueState、ListState、MapState、ReducingState、AggregatingState。
• Operator State（算子状态）
  • 绑定到算子并行实例（SubTask），与 Key 无关。
  • 常用于 Source Connector（记录读取的 Offset）或 Sink Connector（事务控制）。
  • 常用接口：ListState、UnionListState、BroadcastState。

二、状态后端（State Backends）

状态存在哪里？是内存还是磁盘？这由 State Backend 决定。在 Flink 1.13 之后，配置方式简化为以下两种主要模式：

1. HashMapStateBackend (基于内存)

• 存储位置：Java 堆内存（Heap）。
• 特点：读写速度极快（对象直接访问，无序列化开销）。
• 适用场景：状态较小（例如仅仅是简单的 Count 或去重），对延迟极其敏感的场景。
• 缺点：受限于 JVM 堆大小，容易 GC；状态过大时可能 OOM。

2. EmbeddedRocksDBStateBackend (基于磁盘)

• 存储位置：TaskManager 本地磁盘（基于 RocksDB 数据库），内存中只作为缓存（Off-heap）。
• 特点：支持超大状态（TB 级别），不受 JVM 堆限制。
• 适用场景：超大窗口、超长周期的聚合、海量 Key 的去重。
• 缺点：需要序列化/反序列化，读写性能略低于内存版；需要调优 RocksDB 参数。

3. 配置示例

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置状态后端为 RocksDB
env.setStateBackend(new EmbeddedRocksDBStateBackend());

// 配合 Checkpoint 存储路径（存储在本地文件系统）
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink/checkpoints");

三、容错核心：Checkpoint

Checkpoint（检查点）是 Flink 容错机制的灵魂。它是一个全局一致性快照，定期将所有算子的状态持久化到远程存储（如 HDFS）。

1. 核心原理：Barrier 对齐

Flink 使用 Chandy-Lamport 算法 的变体。

1. Barrier 注入：JobManager 向 Source 发送 Checkpoint Barrier。
2. Barrier 流动：Barrier 像普通数据一样在流中传输。
3. 对齐（Alignment）：当算子有多个输入流时，必须等待所有流的 Barrier 到齐，才能进行 Snapshot。这保证了状态的一致性（即 Exactly-Once）。
4. 异步快照：算子将状态写入远程存储（异步过程），不阻塞数据处理。
5. 确认完成：所有算子都完成快照后，JobManager 确认 Checkpoint 成功。

2. Checkpoint 配置实战

默认情况下 Checkpoint 是关闭的，生产环境必须开启。

// 1. 开启 Checkpoint，每 5000ms 触发一次
env.enableCheckpointing(5000);

// 2. 设置 Checkpoint 模式（默认 EXACTLY_ONCE，也可以设为 AT_LEAST_ONCE）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 3. 设置两次 Checkpoint 之间的最小间隔（防止频繁 Checkpoint 导致性能下降）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);

// 4. Checkpoint 超时时间（默认 10分钟）
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 5. 允许同时进行的 Checkpoint 数量（通常设为 1）
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 6. 开启作业取消时保留 Checkpoint（非常重要！否则 Cancel 任务会删除 Checkpoint）
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// 7. 容忍 Checkpoint 失败次数（默认 0，即 Checkpoint 失败会导致任务重启）
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);

四、Savepoint：手动的超级 Checkpoint

虽然 Checkpoint 和 Savepoint 看起来很像（都是快照），但它们的定位完全不同：

常用命令

# 触发 Savepoint
bin/flink savepoint <jobId> [targetDirectory]

# 从 Savepoint 重启作业 (或者 Checkpoint)
bin/flink run -s <savepointPath> ...

五、重启策略（Restart Strategies）

当任务发生故障（Exception）时，Flink 会尝试根据配置的策略自动重启。

// 1. 固定延迟重启（尝试 3 次，每次间隔 10秒）
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3, 
Duration.ofSeconds(10)
));

// 2. 失败率重启（在 5 分钟内失败超过 3 次则停止，否则每次间隔 10秒重启）
env.setRestartStrategy(RestartStrategies.failureRateRestart(
3, 
Duration.ofMinutes(5), 
Duration.ofSeconds(10)
));

// 3. 无重启（直接失败）
env.setRestartStrategy(RestartStrategies.noRestart());

六、总结

• State 是 Flink 实现复杂逻辑的记忆。
• State Backend 决定了记忆存哪里（内存快但小，RocksDB 大但需序列化）。
• Checkpoint 是自动化的定期备份，保证故障恢复后的数据一致性。
• Savepoint 是手动的高级备份，用于版本升级和应用迁移。

掌握了状态与容错，你的 Flink 任务才算真正具备了“生产级”的健壮性。下一篇，我们将探讨 Flink SQL，看看如何用 SQL 解决 80% 的流计算需求。

原文来自：http://blog.daimajiangxin.com.cn

源码地址：https://gitee.com/daimajiangxin/flink-learning