Flink作业跑着跑着突然重启了——Checkpoint到底怎么工作的？

Checkpoint是Flink最核心的容错机制，没有之一。它决定了作业崩溃后能不能恢复、数据会不会丢。我在元宇宙社交App从0搭建Flink实时平台时，Checkpoint是我花时间最多的部分——配置不好作业就频繁重启，配置好了就很稳定。今天把Checkpoint的原理、配置和踩坑彻底讲清楚。

Checkpoint的完整过程

1. JobManager发起： JobManager定期向所有Source算子发送Checkpoint Barrier（一个特殊的标记）。
2. Barrier随数据流动： Barrier像数据一样在算子之间流动。每个算子收到Barrier后，把自己的状态写入State Backend。
3. Barrier对齐： 如果一个算子有多个输入流，它要等所有输入流的Barrier都到齐才做Checkpoint。这就是“Barrier对齐”。
4. 确认完成： 所有算子都完成后，JobManager收到确认，这次Checkpoint标记为成功。

打个比方：Checkpoint就像游戏存档。你每1分钟自动存档一次，死了就从最近的存档点继续。Barrier就像存档时的“暂停键”，确保所有管线的状态都是同一时刻的。

生产环境的Checkpoint配置

// 生产环境推荐配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 开启Checkpoint，每60秒做一次
env.enableCheckpointing(60000);
 
// Exactly-Once语义（默认）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
// Checkpoint超时时间（如果超过10分钟还没完成就放弃）
env.getCheckpointConfig().setCheckpointTimeout(600000);
 
// 最小间隔（上一次Checkpoint完成后至少等30秒再开始下一次）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
 
// 允许1个Checkpoint同时进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
 
// 作业取消时保留Checkpoint（重要！防止手动停作业后状态丢失）
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
 
// State Backend：生产环境用RocksDB（支持大状态）
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints");