一：检查点配置
Flink的配置文件当中有检查点的配置，还有另外一中是在代码里边配置。
二：启动检查点
默认情况下，Flink程序是禁用检查点的。如果想要为Flink应用开启自动保存快照的功能，需要在代码中显式地调用执行环境的.enableCheckpointing()方法：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 每隔1秒启动一次检查点保存 env.enableCheckpointing(1000);这里需要传入一个长整型的毫秒数，表示周期性保存检查点的间隔时间。如果不传参数直接启用检查点，默认的间隔周期为500毫秒，这种方式已经被弃用。
检查点的间隔时间是对处理性能和故障恢复速度的一个权衡。如果我们希望对性能的影响更小，可以调大间隔时间；而如果希望故障重启后迅速赶上实时的数据处理，就需要将间隔时间设小一些。
三：检查点存储
检查点具体的持久化存储位置，取决于“检查点存储”的设置。默认情况下，检查点存储在JobManager的堆内存中。而对于大状态的持久化保存，Flink也提供了在其他存储位置进行保存的接口。
具体可以通过调用检查点配置的.setCheckpointStorage()来配置，需要传入一个CheckpointStorage的实现类。Flink主要提供了两种CheckpointStorage：作业管理器的堆内存和文件系统。
// 配置存储检查点到JobManager堆内存 env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage()); // 配置存储检查点到文件系统 env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));对于实际生产应用，我们一般会将CheckpointStorage配置为高可用的分布式文件系统
四：其它高级配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 启动检查点 当前是10s进行一次恢复 env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);//精准一次，默认是精准一次的。 // 设置状态后端 默认是Hash env.setStateBackend(new HashMapStateBackend()); // 检查点存储 CheckpointConfig checkpointConfig = env.getCheckpointConfig(); // checkpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage()); checkpointConfig.setCheckpointStorage("hdfs://bigdata138:8020/ck"); // 检查点模式(精准一次性) checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 设置超时时间 checkpointConfig.setCheckpointTimeout(6000L);//超时就报错。 // 两个检查点间隔时间 checkpointConfig.setMinPauseBetweenCheckpoints(2000L);//备份时间太长，如果5秒一次，可能一直都在做备份。 歇多长时间再备份。 // 最大并发检查点数量 checkpointConfig.setMaxConcurrentCheckpoints(1);//一般就是一个。这个值不需要去改 // 开启外部持久化存储 当作业取消的时候，检查点是否要保留。 checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION); // 检查点连续失败次数,一般不需要改，一般都是0 checkpointConfig.setTolerableCheckpointFailureNumber(0); // flink提供的另外一种容错手段， 3秒重启一次， env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));//默认重启Interger最大值的次数。 // 开启非对齐检查点 checkpointConfig.enableUnalignedCheckpoints(true);//开启非对齐检查点。 // 对齐检查点超时时间，前提是必须打开非对齐检查点。 checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(10));//先尝试对齐检查点，如果10秒钟对齐检查点没搞定，切换到非对齐检查点。
)
