一、启动脚本分析
1、 WordCount启动命令
bin/flink run examples/streaming/SocketWindowWordCount.jar --hostname localhost --port 9000
2、bin/flink 脚本分析
从上图中我圈的红色部分可以看出，最终调用的是“org.apache.flink.client.cli.CliFrontend”这个类。
二、CliFrontend启动类分析
1、main方法
从下面的代码可以看出，main方法中首先是寻找配置文件路径，然后加载配置文件等等，最后重要的代码就是new 了一个CliFrontend对象出来，然后调用cli.parseAndRun(args)方法；
public static void main(final String[] args) {EnvironmentInformation.logEnvironmentInfo(LOG, "Command Line Client", args);// 1. find the configuration directoryfinal String configurationDirectory = getConfigurationDirectoryFromEnv();// 2. load the global configurationfinal Configuration configuration =GlobalConfiguration.loadConfiguration(configurationDirectory);// 3. load the custom command linesfinal List<CustomCommandLine> customCommandLines =loadCustomCommandLines(configuration, configurationDirectory);int retCode = 31;try {final CliFrontend cli = new CliFrontend(configuration, customCommandLines);SecurityUtils.install(new SecurityConfiguration(cli.configuration));retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.parseAndRun(args));} catch (Throwable t) {final Throwable strippedThrowable =ExceptionUtils.stripException(t, UndeclaredThrowableException.class);LOG.error("Fatal error while running command line interface.", strippedThrowable);strippedThrowable.printStackTrace();} finally {System.exit(retCode);}}
2、cli.parseAndRun(args)方法
这个方法并没有做什么实质性的动作，只是去解析了我们命令行的参数，通过参数选择调用适当方法，比如我们上面的命令输入的是run，后面就会调用run(params)方法；
public int parseAndRun(String[] args) {// check for actionif (args.length < 1) {CliFrontendParser.printHelp(customCommandLines);System.out.println("Please specify an action.");return 1;}// get actionString action = args[0];// remove action from parametersfinal String[] params = Arrays.copyOfRange(args, 1, args.length);try {// do actionswitch (action) {case ACTION_RUN:run(params);return 0;case ACTION_RUN_APPLICATION:runApplication(params);return 0;case ACTION_LIST:list(params);return 0;case ACTION_INFO:info(params);return 0;case ACTION_CANCEL:cancel(params);return 0;case ACTION_STOP:stop(params);return 0;case ACTION_SAVEPOINT:savepoint(params);return 0;case "-h":case "--help":CliFrontendParser.printHelp(customCommandLines);return 0;case "-v":case "--version":String version = EnvironmentInformation.getVersion();String commitID = EnvironmentInformation.getRevisionInformation().commitId;System.out.print("Version: " + version);System.out.println(commitID.equals(EnvironmentInformation.UNKNOWN)? "": ", Commit ID: " + commitID);return 0;default:System.out.printf("\"%s\" is not a valid action.\n", action);System.out.println();System.out.println("Valid actions are \"run\", \"run-application\", \"list\", \"info\", \"savepoint\", \"stop\", or \"cancel\".");System.out.println();System.out.println("Specify the version option (-v or --version) to print Flink version.");System.out.println();System.out.println("Specify the help option (-h or --help) to get help on the command.");return 1;}} catch (CliArgsException ce) {return handleArgException(ce);} catch (ProgramParametrizationException ppe) {return handleParametrizationException(ppe);} catch (ProgramMissingJobException pmje) {return handleMissingJobException();} catch (Exception e) {return handleError(e);}}
3、cli.run(params)方法
在这个方法里面首先去解析命令参数，并判断有没有help，有的话直接输出并返回，然后去解析一些运行参数，获取依赖jar包的信息，并根据以上信息去生成一个有效的配置类，最后调用cli.executeProgram(effectiveConfiguration, program)方法，注意这里会生成一个program，后续的类加载器的使用就是在这里设置的;
protected void run(String[] args) throws Exception {LOG.info("Running 'run' command.");final Options commandOptions = CliFrontendParser.getRunCommandOptions();final CommandLine commandLine = getCommandLine(commandOptions, args, true);// evaluate help flagif (commandLine.hasOption(HELP_OPTION.getOpt())) {CliFrontendParser.printHelpForRun(customCommandLines);return;}final CustomCommandLine activeCommandLine =validateAndGetActiveCommandLine(checkNotNull(commandLine));final ProgramOptions programOptions = ProgramOptions.create(commandLine);final List<URL> jobJars = getJobJarAndDependencies(programOptions);final Configuration effectiveConfiguration =getEffectiveConfiguration(activeCommandLine, commandLine, programOptions, jobJars);LOG.debug("Effective executor configuration: {}", effectiveConfiguration);try (PackagedProgram program = getPackagedProgram(programOptions, effectiveConfiguration)) {executeProgram(effectiveConfiguration, program);}}
4、cli.executeProgram(effectiveConfiguration, program)方法
这个方法就做了一件事，就是去调用ClientUtils.executeProgram(
new DefaultExecutorServiceLoader(), configuration, program, false, false);
protected void executeProgram(final Configuration configuration, final PackagedProgram program)throws ProgramInvocationException {ClientUtils.executeProgram(new DefaultExecutorServiceLoader(), configuration, program, false, false);}
三 ClientUtils工具类分析
1、ClientUtils.executeProgram方法
可以看出分为几个步骤，先是获取用户配置使用的类加载器，然后设置成当前的上下文类加载器，再设置上下文环境，最后通过代理的方式执行用户jar包的main方法；
public static void executeProgram(PipelineExecutorServiceLoader executorServiceLoader,Configuration configuration,PackagedProgram program,boolean enforceSingleJobExecution,boolean suppressSysout)throws ProgramInvocationException {checkNotNull(executorServiceLoader);/***1、获取用户类加载器（这里的用户类加载其实是配置文件里面有个参数可以进行配置（CoreOptions.CLASSLOADER_RESOLVE_ORDER）* 这个参数会决定使用ChildFirstClassLoader还是ParentFirstClassLoader）* 详情请看 buildUserCodeClassLoader方法*/final ClassLoader userCodeClassLoader = program.getUserCodeClassLoader();//2、获取当前上下文类加载器（这里用于后续finally重新设置会此类加载器）final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();try {//3、重新设置当前上下文Thread.currentThread().setContextClassLoader(userCodeClassLoader);LOG.info("Starting program (detached: {})",!configuration.getBoolean(DeploymentOptions.ATTACHED));ContextEnvironment.setAsContext(executorServiceLoader,configuration,userCodeClassLoader,enforceSingleJobExecution,suppressSysout);//设置流处理任务的环境上下文，也就是我们平常使用的StreamExecutionEnvironment.getExecutionEnvironment()获取到的上下文StreamContextEnvironment.setAsContext(executorServiceLoader,configuration,userCodeClassLoader,enforceSingleJobExecution,suppressSysout);try {//执行用户jar包的main方法program.invokeInteractiveModeForExecution();} finally {ContextEnvironment.unsetAsContext();StreamContextEnvironment.unsetAsContext();}} finally {Thread.currentThread().setContextClassLoader(contextClassLoader);}}
如果觉得《【Flink源码分析】Flink 命令启动全流程》对你有帮助，请点赞、收藏，并留下你的观点哦！
