Flink 1.9源码学习02 ----JobManager启动源码分析

ゞ浴缸里的玫瑰 2023-06-05 13:58 29阅读 0赞

在上一篇我们看到jobmanager的启动类org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint

来看一下StandaloneSessionClusterEntrypoint启动类的一些重要的方法：

![20191009114656602.png][]

\--->>>

我们先看main方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

ClusterEntrypoint是一个抽象类，

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

\--->>>

调用它的方法：

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);
       });
    }

\--->>>

启动集群的具体方法：

public void startCluster() throws ClusterEntrypointException {
       LOG.info("Starting {}.", getClass().getSimpleName());
    
       try {
    
          configureFileSystems(configuration);
    
          SecurityContext securityContext = installSecurityContext(configuration);
    
          securityContext.runSecured((Callable<Void>) () -> {
             //启动集群，传入配置文件参数
             runCluster(configuration);
    
             return null;
          });
       } catch (Throwable t) {
          final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
    
          try {
             // clean up any partial state
             shutDownAsync(
                ApplicationStatus.FAILED,
                ExceptionUtils.stringifyException(strippedThrowable),
                false).get(INITIALIZATION_SHUTDOWN_TIMEOUT.toMilliseconds(), TimeUnit.MILLISECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
             strippedThrowable.addSuppressed(e);
          }
    
          throw new ClusterEntrypointException(
             String.format("Failed to initialize the cluster entrypoint %s.", getClass().getSimpleName()),
             strippedThrowable);
       }
    }

\--->>>

我们看看 runCluster这个方法做了啥子：

private void runCluster(Configuration configuration) throws Exception {
       synchronized (lock) {
    
          //todo 初始化了一些ClusterEntrypoint.java中的一些服务,比如：HA,blob,heartbeat,metricRegistry这些
          initializeServices(configuration);
    
          //todo 将host信息写入配置
          configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
          configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());
    
          final DispatcherResourceManagerComponentFactory<?> dispatcherResourceManagerComponentFactory = createDispatcherResourceManagerComponentFactory(configuration);
    
          //todo 将工厂真正开启接口，其中包括了一些创建以及启动ResourceManager（有用于请求solt的RPC，初始化所有solt到resourceManager的soltManager的RPC(这个会在jobmanager接收到jobGraph后调用),TM心跳等），启动web服务
          clusterComponent = dispatcherResourceManagerComponentFactory.create(
             configuration,
             commonRpcService,
             haServices,
             blobServer,
             heartbeatServices,
             metricRegistry,
             archivedExecutionGraphStore,
             new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()),
             this);
    
          clusterComponent.getShutDownFuture().whenComplete(
             (ApplicationStatus applicationStatus, Throwable throwable) -> {
                //todo 出错了，关闭方法
                if (throwable != null) {
                   shutDownAsync(
                      ApplicationStatus.UNKNOWN,
                      ExceptionUtils.stringifyException(throwable),
                      false);
                } else {
                   //todo 这是一般的关闭路径。如果是单独的更具体的关闭
                   //todo 已经触发，这将什么也不做
                   shutDownAsync(
                      applicationStatus,
                      null,
                      true);
                }
             });
       }
    }

\--->>>

具体的ResourceManager的初始化：

点进去：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

这是一个接口：

DispatcherResourceManagerComponentFactory

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

我们看到抽象类

AbstractDispatcherResourceManagerComponentFactory 实现了这个接口

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

创建resourceManager 用于接收slot的请求：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

我们发现它是一个接口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6][]

来看类：

package org.apache.flink.runtime.resourcemanager;

这是一个枚举类：

public enum StandaloneResourceManagerFactory implements ResourceManagerFactory<ResourceID> {

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8][]

点进去看看发现这个类创建了一个继承了resourceManager 这个抽象类

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9][]

Resourcemanager 这个抽象类：

它

继承ResourceIDRetrievable>
          继承FencedRpcEndpoint<ResourceManagerId>
          实现 ResourceManagerGateway, LeaderContender {

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10][]

这个接口下的几个重要的RPC方法具体实现：

1）向resourceManager请求slot：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11][]

2）这个rpc想resourceManager发送包括像taskManagaer有多少可分配的solt,哪些已分配的solt,solt的状态等

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12][]

\--->>>

最后创建完毕之后调用启动方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13][]

调用的是rpcServer

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14][]

\--->>>

完事之后开始调度了：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15][]

\--->>>

创建了一个Dispatcher调度对象

看下Dispatcher是用来干嘛的（StandaloneDispatcher都是调用了父类的初始化方法super()创建一个Dispatcher.java对象）

来看一下Dispatcher实现了什么接口(ResourceManager同理)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16][]

看它实现的接口：

里面有提交job：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17][]

ok，我们再返回去找到具体的实现方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 18][]

点进去：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 19][]

最后：

他实现了submitJob()接口用于启动一个RPC,接受参数可以看到接受到一个JobGraph，这就意味着这和job任务启动有关，后面随缘更新到job启动Graph转换会提到

回到前面的Dispatcher.start()将传入的rpcService启动起来了,等待接受来自Driver端提交上来的JobGraph差不多启动完成了

参考：[https://www.cnblogs.com/ljygz/p/11405572.html][https_www.cnblogs.com_ljygz_p_11405572.html]

[20191009114656602.png]: /images/20230601/07253a1dcba349d9a2b71ab6257f0d7b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20230601/646edd142ae24718a2187646c501c84b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20230601/253c71f5e9d74ffba066781baf7b291c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20230601/e8bc97e6c11e466f825a175aeebe2249.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20230601/c92ba39b972c4ca2af6b6a5a11a7f097.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20230601/e8b90a36377945819f323b7f866b394d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20230601/cedcf66ee8454aca870bb196396e4b69.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6]: /images/20230601/88221b7c9d7945248fdab2d84599d05e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7]: /images/20230601/9a5c6e00c2f04418a0979bd567956987.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8]: /images/20230601/f08c4456dd924eb38f03646fd3961fd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9]: /images/20230601/a37c90823167414ca3c2a8b603d715b5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10]: /images/20230601/175233d5a0a74df2a6157f63a4cddbd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11]: /images/20230601/06acc3f9d9e94c2fa25d15e526011a91.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12]: /images/20230601/37427bb966464442810e07b3cc859940.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 13]: /images/20230601/66f78cae83fe464e809ec8247bae8208.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 14]: /images/20230601/bb091b7e67154f58a267ee902d1e4ab8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 15]: /images/20230601/a757dfa601d14c078f20a5a60a7b6e07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 16]: /images/20230601/519b465d91da42fdaddbcd67217ff26b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 17]: /images/20230601/5548f075d8134d51ac0660f56b0e257e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 18]: /images/20230601/cf41f49eaea84df99743b6df780fb3f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 19]: /images/20230601/57f9d8723bac4e638376d86f225a0628.png
[https_www.cnblogs.com_ljygz_p_11405572.html]: https://www.cnblogs.com/ljygz/p/11405572.html