【Flink】Flink与Nacos集成

超、凢脫俗 2023-01-05 12:01 237阅读 0赞

## 1. 集成原因 ##

在传统的开发中，我们通常将系统配置保存在数据库中，供程序访问；在微服务中，我们通常将系统配置保存在分布式配置中心，以便程序能够在运行时动态的获取配置而不用重启程序。同样的，在Flink程序中，我们也可以将配置存储在固定的配置中心进行统一的管理。

Flink本身提供了ParameterTool工具类来从启动参数、properties文件以及系统环境变量等位置获取配置信息，但由于ParameterTool依赖Flink的任务节点，必须在每个任务节点上都保存一份配置，很不方便，所以我们需要把配置保存在数据库或者配置中心通过REST API来获取配置数据。

本文就提供一种Flink和Nacos的集成方式，通过Nacos来动态获取配置程序的公共配置。

## 2. 集成方式 ##

Nacos作为微服务中常用的分布式配置中心和注册中心，提供了配置和注册服务的能力。本次只需要使用到配置中心，能够获取到配置即可，所以所需依赖如下：

<dependency>xml
    	<groupId>com.alibaba.nacos</groupId>
    	<artifactId>nacos-client</artifactId>
    	<version>1.2.1</version>
    </dependency>

然后新建一个工具类来获取Nacos配置：

/** * flink nacos support * * @author wxg */
    public class FlinkNacosConfig { 
        private static final Logger logger = LoggerFactory.getLogger(FlinkNacosConfig.class);
        private static final String ENABLE_NACOS_KEY = "sim.flink.nacos.enable";
        private static final String SERVER_ADDR_KEY = "sim.flink.nacos.serverAddr";
        private static final String NAMESPACE_KEY = "sim.flink.nacos.namespace";
        private static final String GROUPS_KEY = "sim.flink.nacos.groups";
        private static final String DATA_IDS_KEY = "sim.flink.nacos.dataIds";
        private static final String TIMEOUT_KEY = "sim.flink.nacos.timeout";
    
        private static final String DEFAULT_GROUP = "DEFAULT_GROUP";
        private static final String DEFAULT_DATA_ID = "appconfig.properties";
        private static final String DEFAULT_TIMEOUT = "3000";
    
        private static String SERVER_ADDR;
        private static String NAMESPACE;
        private static String[] GROUPS;
        private static String[] DATA_IDS;
        private static Integer TIMEOUT;
    
        private static Properties CONFIG_MAP = new Properties();
        private static Properties NACOS_CONFIG_MAP = new Properties();
    
        private static final String DEFAULT_CONFIG_FILE = "application.properties";
    
        static { 
            InputStream inputStream = FlinkNacosConfig.class.getClassLoader().getResourceAsStream(DEFAULT_CONFIG_FILE);
            if (inputStream == null) { 
                logger.info("no flink nacos config found");
            } else { 
                try { 
                    CONFIG_MAP.load(inputStream);
                    String enableNacos = CONFIG_MAP.getProperty(ENABLE_NACOS_KEY, "false");
                    if (Boolean.valueOf(enableNacos)) { 
                        SERVER_ADDR = CONFIG_MAP.getProperty(SERVER_ADDR_KEY);
                        NAMESPACE = CONFIG_MAP.getProperty(NAMESPACE_KEY);
                        GROUPS = CONFIG_MAP.getProperty(GROUPS_KEY, DEFAULT_GROUP).split(",");
                        DATA_IDS = CONFIG_MAP.getProperty(DATA_IDS_KEY, DEFAULT_DATA_ID).split(",");
                        TIMEOUT = Integer.valueOf(CONFIG_MAP.getProperty(TIMEOUT_KEY, DEFAULT_TIMEOUT));
                        initNacos();
                    } else { 
                        logger.debug("Naocs config is not enable");
                    }
                } catch (IOException e) { 
                    e.printStackTrace();
                    logger.error("", e);
                }
    
            }
        }
    
        private static void initNacos() { 
            Properties properties = new Properties();
            properties.put("serverAddr", SERVER_ADDR);
            properties.put("namespace", NAMESPACE);
    
    
            String content = null;
            try { 
                logger.info("loading config from nacos...");
    
                ConfigService configService = NacosFactory.createConfigService(properties);
                for (String group : GROUPS) { 
                    for (String dataId : DATA_IDS) { 
                        content = configService.getConfig(dataId, group, TIMEOUT);
                        if (content != null) { 
                            NACOS_CONFIG_MAP.load(new StringReader(content));
                            configService.addListener(dataId, group, new PropertiesListener() { 
                                @Override
                                public void innerReceive(Properties properties) { 
                                    logger.info("config file {}:{} refreshed", group, dataId);
                                    NACOS_CONFIG_MAP.putAll(properties);
                                }
                            });
                        }
                    }
                }
                logger.info("load config from nacos success");
    
            } catch (NacosException | IOException e) { 
                e.printStackTrace();
                logger.error("", e);
            }
    
        }
    
        public static String getConfigValue(String key) { 
            return NACOS_CONFIG_MAP.getProperty(key);
        }
    
        public static String getConfigValue(String key, String defaultValue) { 
            return NACOS_CONFIG_MAP.getProperty(key, defaultValue);
        }
    }

## 3. 使用方式 ##

String configValue = FlinkNacosConfig.getConfigValue(your_config_key);

配置文件：

# 是否开启nacos支持
    sim.flink.nacos.enable=true
    # nacos地址
    sim.flink.nacos.serverAddr=
    # nacos命名空间
    sim.flink.nacos.namespace=
    # nacos配置分组 
    sim.flink.nacos.groups=
    # nacos的data-id
    sim.flink.nacos.dataIds=

当配置修改时，将会自动获取到最新的配置内容。