Flink之Task重启策略-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink之Task重启策略。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Task重启策略

1 策略API

noRestart

无参数,task失败后不重启,整个job同时失败,默认策略.

代码示例文章来源地址https://www.toymoban.com/news/detail-777452.html
```
RestartStrategies.noRestart();
```
fixedDelayRestart

参数注释

restartAttempts 最大重启次数

delayBetweenAttempts 重启时间间隔

代码示例
```
// 最多重启5次,每次任务失败后间隔1s重启
RestartStrategies.fixedDelayRestart(5, 1000);
```

参数	注释
restartAttempts	最大重启次数
delayBetweenAttempts	重启时间间隔

exponentialDelayRestart

参数	注释
initialBackoff	重启间隔惩罚时长初始值(重启延迟时间)
maxBackoff	重启间隔最大惩罚时长
backoffMultiplier	重启间隔时长的惩罚倍数
resetBackoffThreshold	重置惩罚时长的平稳运行时长(平稳运行时长达到这个阈值后,再次发生故障则重启延迟时间恢复到初始值)
jitterFactor	取一个随机数,来加在重启时间点上,已让每次重启的时间呈现一定随机性(避免某一时刻集群中有大量的task同时重启,如果task重启时间是规律性的就可能发生这种情况)

代码示例

// 第一次失败后间隔1s重启任务,如果稳定运行时长没有达到120s就发生task失败,则重启间隔时长=上一次重启间隔时长*1.2,如果稳定运行时长超过120秒则重启间隔时长恢复到1s
RestartStrategies.exponentialDelayRestart(Time.seconds(1), Time.seconds(30), 1.2, Time.seconds(120), 0.56);

failureRateRestart

参数	注释
failureRate	指定时间范围内的最大Task任务失败率(次数)
failureInterval	指定时间范围
delayInterval	重启时间间隔

代码示例

// task失败重启间隔为1s,只要在30分钟内task失败重启次数没超过3次就可以一直执行这个策略,如果超过则job停止
RestartStrategies.failureRateRestart(3, Time.minutes(30), Time.seconds(1));

fallBackRestart

无参数,常用于自定义的RestartStrategy,即用户自定义了重启策略,且将其配置在了flink-conf.yaml文件中,也就是说调用这个方法时会读取集群的配置文件,根据配置文件的内容调整策略

代码示例
```
RestartStrategies.fallBackRestart();
```

2 代码详情

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

        // 开启并设置checkpoint的时间间隔
        env.enableCheckpointing(3000);
        // 设置checkpoint的存储位置
        env.getCheckpointConfig().setCheckpointStorage(new Path("hdfs://lx01:8020/flink-ck"));
        // 允许checkpoint失败的最大次数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        // checkpoint的算法模式,是否需要对其(EXACTLY_ONCE)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // job取消是否保留checkpoint数据
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置checkpoint对齐的超时时间
        env.getCheckpointConfig().setAlignedCheckpointTimeout(Duration.ofMillis(10000));
        // 两次checkpoint的最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        // 并行最大的checkpoint数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(3);
        // 选择后端状态(默认HashMapStateBackend)
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        // TODO Task重启策略
        RestartStrategies.RestartStrategyConfiguration restartStrategy = null;
        // 第一次失败后间隔1s重启任务,如果稳定运行时长没有达到120s就发生task失败,则重启间隔时长=上一次重启间隔时长*1.2,如果稳定运行时长超过120秒则重启间隔时长恢复到1s
        restartStrategy = RestartStrategies.exponentialDelayRestart(Time.seconds(1), Time.seconds(30), 1.2, Time.seconds(120), 0.56);

        // 配置Task重启策略
        env.setRestartStrategy(restartStrategy);
      
        // ...业务代码
      
        env.execute();
    }
}