Flink学习12容错机制（二）重启策略实例

示例代码如下&＃xff1a;

1. 首先要开启checkpointing, 如   env.enableCheckpointing(5000);

2. 选择一种重启策略

//固定间隔重启&＃xff1a;最多重启五次&＃xff0c;重启间隔2000毫秒env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5,2000));
//失败率&＃xff1a;failureRate是每个测量时间间隔最大失败次数
//第二个参数failureInterval失败率测量的时间间隔; //第三个参数delayInterval是两次连续重启尝试的时间间隔env.setRestartStrategy(RestartStrategies.failureRateRestart(5,Time.of(5, TimeUnit.MINUTES),Time.of(10, TimeUnit.SECONDS)));
//无重启env.setRestartStrategy(RestartStrategies.noRestart());


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.util.concurrent.TimeUnit;public class RestartStrategyReview {public static void main(String[] args) throws Exception{//1.获取flink流计算的运行环境StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();//2. 开启checkpointing 只有开启了才会有重启策略&＃xff0c;参数为周期&＃xff08;毫秒&＃xff09;env.enableCheckpointing(5000);//3. 设置重启策略//3.1 固定间隔重启&＃xff1a;最多重启五次&＃xff0c;重启间隔2000毫秒env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5,2000));//3.2 失败率&＃xff1a;failureRate是每个测量时间间隔最大失败次数//第二个参数failureInterval失败率测量的时间间隔; //第三个参数delayInterval是两次连续重启尝试的时间间隔
// env.setRestartStrategy(RestartStrategies.failureRateRestart(5,
// Time.of(5, TimeUnit.MINUTES),Time.of(10, TimeUnit.SECONDS)));//3.3 无重启
// env.setRestartStrategy(RestartStrategies.noRestart());//4. 连接数据源DataStreamSource lines &＃61; env.socketTextStream("192.168.***.****", 8888);//5. Transformation 数据转换SingleOutputStreamOperator> wordAndOne &＃61; lines.flatMap(new FlatMapFunction>() {&＃64;Overridepublic void flatMap(String line, Collector> collector) throws Exception {String[] words &＃61; line.split(",");for (int i &＃61; 0; i > summed &＃61; wordAndOne.keyBy(0).sum(1);//6. sink 打印输出summed.print();//7. 启动流计算env.execute("RestartStrategyReview");}
}

在flink输出次数为5后&＃xff0c;输入error&＃xff0c;报错重启&＃xff0c;再次输入flink后统计次数是6.