Flink重启策略机制

restart-strategy: failure-rate
restart-strategy.failure-rate.max-failures-per-interval: 3 #失败的次数
restart-strategy.failure-rate.failure-rate-interval: 2 min #时间段内
restart-strategy.failure-rate.delay: 5s #重试的间隔时间

restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 3 #次数
restart-strategy.fixed-delay.delay: 10 s #间隔

env.setRestartStrategy(RestartStrategies.failureRateRestart(
3,//最大失败次数
Time.of(2, TimeUnit.MINUTES), // 衡量失败次数的是时间段
Time.of(5, TimeUnit.SECONDS) // 间隔
));

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
3,,// 尝试重启的次数
Time.of(5, TimeUnit.SECONDS) // 间隔
));

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart());

//获取flink的运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】
env.enableCheckpointing(1000);
// 高级选项：
// 设置模式为exactly-once （这是默认值）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 同一时间只允许进行一个检查点
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
//ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
//ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);