flink启动原理入口分析

　　此处使用分析的flink版本为1.10.0
　　直接使用jobmanager.sh 和 taskmanager.sh 启动单机模式的flink服务的脚本逻辑其实是调用 flink-daemon.sh
　　Usage: flink-daemon.sh (start|stop|stop-all) (taskexecutor|zookeeper|historyserver|standalonesession|standalonejob) [args]
　　里面有启动的入口类，即可以追踪到flink单机版启动逻辑 case $DAEMON in     (taskexecutor)         CLASS_TO_RUN=org.apache.flink.runtime.taskexecutor.TaskManagerRunner     ;;      (zookeeper)         CLASS_TO_RUN=org.apache.flink.runtime.zookeeper.FlinkZooKeeperQuorumPeer     ;;      (historyserver)         CLASS_TO_RUN=org.apache.flink.runtime.webmonitor.history.HistoryServer     ;;      (standalonesession)         CLASS_TO_RUN=org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint     ;;      (standalonejob)         CLASS_TO_RUN=org.apache.flink.container.entrypoint.StandaloneJobClusterEntryPoint     ;;
　　可以得出jobmanager和taskmanager的独立启动入口类
　　org.apache.flink.container.entrypoint.StandaloneJobClusterEntryPoint
　　org.apache.flink.runtime.taskexecutor.TaskManagerRunner
　　和本地集群方式启动入口类
　　org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint
　　其他脚本的执行其实都是使用的这个脚本逻辑，如start-cluster.sh脚本启动，不管是启动HA模式，还是非HA模式，都会执行jobmanager.sh脚本，所以都会走flink-daemon.sh
　　那我们就可以通过代码进行分析了
　　StandaloneSessionClusterEntrypoint启动源码，StandaloneSessionClusterEntrypoint继承自SessionClusterEntrypoint // 进行启动检查，获取启动环境信息，如版本，scala版本，git提交号，jvm版本，hadoop版本，javahome等 EnvironmentInformation.logEnvironmentInfo(LOG, StandaloneSessionClusterEntrypoint.class.getSimpleName(), args); // 捕捉系统信号windows包含：终止（TERM），INT（键盘中断）；其它系统多包含：HUP（终端挂起或者控制进程终止） SignalHandler.register(LOG); // 注册JVM关闭的钩子函数 设置5秒延迟退出 JvmShutdownSafeguard.installAsShutdownHook(LOG); // 获取启动行参数，使用的是org.apache.commons.cli // 所有的启动参数都写在了org.apache.flink.runtime.entrypoint.parser.CommandLineOptions类中 // 包括： // c configDir: Directory which contains the configuration file flink-conf.yml. // r webui-port: Port for the rest endpoint and the web UI. // D property=value: use value for given property // h host: Hostname for the RPC service. // x executionMode: Deprecated option EntrypointClusterConfiguration entrypointClusterConfiguration = null; final CommandLineParser commandLineParser = new CommandLineParser<>(new EntrypointClusterConfigurationParserFactory());  try {    entrypointClusterConfiguration = commandLineParser.parse(args); } catch (FlinkParseException e) {    LOG.error(＂Could not parse command line arguments {}.＂, args, e);    commandLineParser.printHelp(StandaloneSessionClusterEntrypoint.class.getSimpleName());    System.exit(1); } // 获取到参数 Configuration configuration = loadConfiguration(entrypointClusterConfiguration);  StandaloneSessionClusterEntrypoint entrypoint = new StandaloneSessionClusterEntrypoint(configuration); // 运行 执行父类方法 ClusterEntrypoint.runClusterEntrypoint(entrypoint);
　　SessionClusterEntrypoint 继承ClusterEntrypoint 	public static void runClusterEntrypoint(ClusterEntrypoint clusterEntrypoint) {  		final String clusterEntrypointName = clusterEntrypoint.getClass().getSimpleName(); 		try { //启动集群方法 			clusterEntrypoint.startCluster(); 		} catch (ClusterEntrypointException e) { 			LOG.error(String.format(＂Could not start cluster entrypoint %s.＂, clusterEntrypointName), e); 			System.exit(STARTUP_FAILURE_RETURN_CODE); 		}  		clusterEntrypoint.getTerminationFuture().whenComplete((applicationStatus, throwable) -> { 			final int returnCode;  			if (throwable != null) { 				returnCode = RUNTIME_FAILURE_RETURN_CODE; 			} else { 				returnCode = applicationStatus.processExitCode(); 			}  			LOG.info(＂Terminating cluster entrypoint process {} with exit code {}.＂, clusterEntrypointName, returnCode, throwable); 			System.exit(returnCode); 		}); 	}
　　进入ClusterEntrypoint的startCluster方法
　　public void startCluster() throws ClusterEntrypointException { 		try { 			replaceGracefulExitWithHaltIfConfigured(configuration); 			// 初始化共享文件系统设置，会将文件系统映射为url 			configureFileSystems(configuration); 			// 初始化安全上下文：其中为进程范围的安全配置，使用可用的安全模块（即Hadoop、JAAS）应用配置。 			SecurityContext securityContext = installSecurityContext(configuration); 			securityContext.runSecured((Callable) () -> { 				runCluster(configuration); 				return null; 			}); 		} 	}
　　接下来进入runCluster方法 	private void runCluster(Configuration configuration) throws Exception { 		synchronized (lock) { 			//初始化集群服务, 			initializeServices(configuration);  			// write host information into configuration 			configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress()); 			configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());  			final DispatcherResourceManagerComponentFactory dispatcherResourceManagerComponentFactory = createDispatcherResourceManagerComponentFactory(configuration);  			clusterComponent = dispatcherResourceManagerComponentFactory.create( 				configuration, 				ioExecutor, 				commonRpcService, 				haServices, 				blobServer, 				heartbeatServices, 				metricRegistry, 				archivedExecutionGraphStore, 				new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()), 				this);
　　根据上述服务创建资源，调度，监控进程DispatcherResourceManagerComponent，并设置同步状态关闭，这之中还包含网关及查询服务检索器的创建，基本了解了flink启动读取运行参数，获取本地配置，并增加了JVM关闭钩子等一些我们平常开发不常用的一些方法，flink的启动流程刚刚迈入门槛。
　　后续看看flink是如何实现这些服务，并管理好task，保证服务的稳定性的。