1. 开发目的

在日常的Flink中开发中，基本是在自己电脑的idea工具上进行Flink程序开发，并在本地联通测试环境进行调试（比如测试环境的Kafka等），当在本地调试通过后将代码打包，然后提交到正式环境运行。但在此过程中，因为本地调试和线上运行时使用的配置不同，会造成我们代码上线时需要修改较多配置（比如Checkpoint地址等），并且还不能修改错，不然可能会污染线上数据。由此开发了一个通用的ModelUtil工具类，可以根据不同的运行环境来决定不同的配置，这样上线时就不需要去特意配置了。

除了本地调试和上线运行外，在我们编写Flink代码时，针对环境变量由较多通用配置，如果每个程序主类都配置一遍，会感觉很麻烦，由此想到，可以将所有通用配置环境放到一个类里面统一配置，根据输入的参数来微调，这样创建Flink执行环境就较为简单了。

2. 环境依赖

具体环境依赖跟Flink依赖一致，可以参考博主另一篇文章： Flink（8）：Flink的API说明和pom文件汇总

3. 具体代码

博主使用的环境为华为云，所以线上的checkpoint保存地址为华为云的obs文件系统，另外博主使用的Flink版本为1.10版本

该工具类包括如下4个功能（方法）：

• 根据传入key获取对应配置文件中的配置值
• 对传入的Flink的流的执行环境配置Checkpoint（filesystem类型）
• 对传入的Flink的流的执行环境配置Checkpoint（rocksdb类型）
• 对传入的Flink的流的执行环境配置重启策略

import org.apache.commons.lang3.SystemUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.contrib.streaming.state.PredefinedOptions;
import org.apache.flink.contrib.streaming.state.RocksDBOptionsFactory;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.TernaryBoolean;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.ResourceBundle;
import java.util.concurrent.TimeUnit;

/**
 * @date: 2021/6/9
 *  @Author ddkk.com  弟弟快看，程序员编程资料站
 * @desc: 模块工具类
 */
public class ModelUtil {

    public static ResourceBundle localConfig = ResourceBundle.getBundle("localConfig");
    public static ResourceBundle config = ResourceBundle.getBundle("config");

    public static Logger logger = LoggerFactory.getLogger(ModelUtil.class);

    /**
     * 根据key获取配置值
     *
     * @param key 配置参数的key
     * @return 配置参数的value
     */
    public static String getConfigValue(String key) {
        if (SystemUtils.IS_OS_WINDOWS) {
            return localConfig.getString(key);
        } else {
            return config.getString(key);
        }
    }

    /**
     * 对传入的Flink的流的执行环境配置Checkpoint
     *
     * @param env             Flink的流的执行环境
     * @param applicationName 应用程序名，会在checkpoint的路径下创建该应用程序的文件夹，用来保存该应用程序的checkpoint
     * @param interval        checkpoint的时间间隔，单位：毫秒，filesystem模式的checkpoint建议间隔为 10s - 60s 之间的整10数
     */
    public static void deployFsCheckpoint(StreamExecutionEnvironment env, String applicationName, long interval) {

        // 启动checkpoint，设置为精确一次，并通过传入的参数设置时间间隔
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointInterval(interval);

        // 设置 状态后端 和 checkpoint 为 filesystem 的模式，并通过配置文件指定文件夹（判断是本地调试还是集群环境）
        String checkpointPath = null;
        if (SystemUtils.IS_OS_WINDOWS) {
            checkpointPath = ModelUtil.getConfigValue("fs.checkpoint.path") + applicationName + "\\\\";
        } else {
            String obsPath = "obs://" +
                    ModelUtil.getConfigValue("obs.ak") + ":" +
                    ModelUtil.getConfigValue("obs.sk") + "@" +
                    ModelUtil.getConfigValue("obs.endpoint") +
                    ModelUtil.getConfigValue("fs.checkpoint.path");
            checkpointPath = obsPath + applicationName + "/";
        }
        env.setStateBackend(new FsStateBackend(checkpointPath));

        // 设置2个checkpoint之间的最小间隔，不需要设置，默认为0
        // env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10 * 1000);

        // 设置能容忍100个检查点的失败
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(100);

        // 当作业被cancel时，不删除外部保存的检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 当在设置的时间内还没有保存成功认为该检查点失败，设置为interval的10倍
        env.getCheckpointConfig().setCheckpointTimeout(interval * 10);

        // 设置同时可以进行10个checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(10);

        logger.info(">>>>> 正在进行环境设置，会创建fs的checkpoint环境，applicationName：" + applicationName + " ; 间隔时间interval：" + interval + " ; ");

    }

    /**
     * 对传入的Flink的流的执行环境配置Checkpoint
     *
     * @param env             Flink的流的执行环境
     * @param applicationName 应用程序名，会在checkpoint的路径下创建该应用程序的文件夹，用来保存该应用程序的checkpoint
     * @param interval        checkpoint的时间间隔，单位：毫秒，RocksDB模式的checkpoint建议间隔为 1分钟到30分钟 之间的整分钟数
     */
    public static void deployRocksdbCheckpoint(StreamExecutionEnvironment env, String applicationName, long interval) {

        // 启动checkpoint，设置为精确一次，并通过传入的参数设置时间间隔
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointInterval(interval);

        // 设置 状态后端 和 checkpoint 为 filesystem 的模式，并通过配置文件指定文件夹（判断是本地调试还是集群环境）
        String checkpointPath = null;
        if (SystemUtils.IS_OS_WINDOWS) {
            checkpointPath = ModelUtil.getConfigValue("rocksdb.checkpoint.path") + applicationName + "\\\\";
        } else {
            String obsPath = "obs://" +
                    ModelUtil.getConfigValue("obs.ak") + ":" +
                    ModelUtil.getConfigValue("obs.sk") + "@" +
                    ModelUtil.getConfigValue("obs.endpoint") +
                    ModelUtil.getConfigValue("rocksdb.checkpoint.path");
            checkpointPath = obsPath + applicationName + "/";
        }
        RocksDBStateBackend rocksDbBackend = new RocksDBStateBackend(new FsStateBackend(checkpointPath), TernaryBoolean.TRUE);
        // 预定义选项，SPINNING_DISK_OPTIMIZED为基于磁盘的优化，一般使用SPINING_DISK_OPTIMIZED_HIGH_MEM，但这会消耗比较多的内存
        rocksDbBackend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM);
        rocksDbBackend.setRocksDBOptions(new RocksDBOptionsFactory() {
            @Override
            public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
                return currentOptions
                        // 指定信息日志文件的最大大小。 如果当前日志文件大于' max_log_file_size '，一个新的信息日志文件将被创建。如果为0，所有日志将被写入一个日志文件。
                        .setMaxLogFileSize(64 * 1024 * 1024)
                        // 信息日志文件的最大保留个数。
                        // .setKeepLogFileNum(3)
                        ;
            }

            @Override
            public ColumnFamilyOptions createColumnOptions(ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
                return currentOptions;
            }
        });
        env.setStateBackend(rocksDbBackend);

        // 设置2个checkpoint之间的最小间隔，不需要设置，默认为0
        // env.getCheckpointConfig().setMinPauseBetweenCheckpoints(interval);

        // 设置能容忍100个检查点的失败
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(100);

        // 当作业被cancel时，不删除外部保存的检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 当在设置的时间内还没有保存成功认为该检查点失败，设置为interval的10倍
        env.getCheckpointConfig().setCheckpointTimeout(interval * 10);

        // 设置同时可以进行10个checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(10);

        logger.info(">>>>> 正在进行环境设置，会创建rocksdb的checkpoint环境，applicationName：" + applicationName + " ; 间隔时间interval：" + interval + " ; ");
    }

    /**
     * 对传入的Flink的流的执行环境配置重启策略
     *
     * @param env Flink的流的执行环境
     */
    public static void deployRestartStrategy(StreamExecutionEnvironment env) {

        // 当任务中异常失败后，会重启任务3次，间隔时间为60秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(60, TimeUnit.SECONDS)));
        logger.info(">>>>> 正在进行环境设置，重启策略为：重启任务3次，间隔时间为60秒");

        // 10分钟内重启5次,每次间隔2分钟（排除了网络等问题，如果再失败，需要手动查明原因）
        // env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(10, TimeUnit.MINUTES), Time.of(2, TimeUnit.MINUTES)));
        // logger.info(">>>>> 正在进行环境设置，重启策略为：10分钟内重启5次,每次间隔2分钟");

    }

}

4. 具体使用

如下代码所示，在具体使用中，通过简短的4行代码既可以配置出Flink的运行环境：

public static void main(String[] args) throws Exception {

    // 包括但不限于kafka的消费者id
    String applicationName = "应用名";

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    ModelUtil.deployRocksdbCheckpoint(env, applicationName, 3 * 1000);
    ModelUtil.deployRestartStrategy(env);

    // 具体业务代码

    env.execute(applicationName);
}

在resources文件夹下，需要配置出本地运行环境和线上运行环境的不同的配置选项，具体文件如下图所示：

线上运行配置如下图所示：

本地运行配置如下图所示：