• Flink 1.13 源码解析——JobManager启动流程之Dispatcher启动

    ​点击这里查看 Flink 1.13 源码解析 目录汇总

    点击查看相关章节：Flink 1.13 源码解析——JobManager启动流程概览

    点击查看相关章节：Flink 1.13 源码解析——JobManager启动流程 WebMonitorEndpoint启动

    点击查看相关章节：Flink 1.13 源码解析——JobManager启动流程之ResourceManager启动

    目录

    一、前言：

    二、DispatcherRunner启动流程

    2.1、DispatcherRunner的Leader选举

    2.2、开始准备构建Dispatcher

    2.2.1、启动JobGraphStore

    2.2.2、寻找中断的Job

    2.2.3、构建Dispatcher并启动

    三、总结

    一、前言：

            在之前的章节里，我们分析了Flink主节点（逻辑JobManager）的启动过程，包括了8个基础环境的创建，核心实例工厂类的创建，以及通过工厂类构建并启动WebMonitorEndpoint、ResourceManager的过程，在这一节中我们来看最后的一部分Dispatcher的启动流程，当然在此之前还是先来复习一下JobManager的一些重要概念以及Dispatcher组件的功能是什么。

            关于Flink的主节点JobManager，他只是一个逻辑上的主节点，针对不同的部署模式，主节点的实现类也不同。

            JobManager（逻辑）有三大核心内容，分别为ResourceManager、Dispatcher和WebmonitorEndpoin：

    ResourceManager：

            Flink集群的资源管理器，只有一个，关于Slot的管理和申请等工作，都有它负责

    Dispatcher：

            1、负责接收用户提交的JobGraph，然后启动一个JobMaster，类似于Yarn中的AppMaster和Spark中的Driver。

            2、内有一个持久服务：JobGraphStore，负责存储JobGraph。当构建执行图或物理执行图时主节点宕机并恢复，则可以从这里重新拉取作业JobGraph

    WebMonitorEndpoint：

            Rest服务，内部有一个Netty服务，客户端的所有请求都由该组件接收处理

    用一个例子来描述这三个组件的功能：

            当Client提交一个Job到集群时（Client会把Job构建成一个JobGraph），主节点接收到提交的job的Rest请求后，WebMonitorEndpoint 会通过Router进行解析找到对应的Handler来执行处理，处理完毕后交由Dispatcher，Dispatcher负责大气JobMaster来负责这个Job内部的Task的部署执行，执行Task所需的资源，JobMaster向ResourceManager申请。 

    二、DispatcherRunner启动流程

            Dispatcher的初始化构成与之前的WebMonitorEndpoint和ResourceManager稍有不同，在构建核心工厂类后，Dispatcher并没有像WebMonitorEndpoint和ResourceManager一样直接构建实例，而是构建了一个DispatcherRunner，并在内部构建了Dispatcher实例并启动。我们来看它是如何实现的，首先还是来到dispatcherResourceManagerComponentFactory.create()方法：

    1. /*
    2. TODO 在该代码的内部会创建Dispatcher组件，并调用start() 方法启动
    3.  */
    4. dispatcherRunner =
    5.         dispatcherRunnerFactory.createDispatcherRunner(
    6.                 highAvailabilityServices.getDispatcherLeaderElectionService(),
    7.                 fatalErrorHandler,
    8.                 new HaServicesJobGraphStoreFactory(highAvailabilityServices),
    9.                 ioExecutor,
    10.                 rpcService,
    11.                 partialDispatcherServices);

    可以看到，这里并没有构建Dispatcher，也没有启动Dispatcher，我们进入createDispatcherRunner方法

    1.     @Override
    2.     public DispatcherRunner createDispatcherRunner(
    3.             LeaderElectionService leaderElectionService,
    4.             FatalErrorHandler fatalErrorHandler,
    5.             JobGraphStoreFactory jobGraphStoreFactory,
    6.             Executor ioExecutor,
    7.             RpcService rpcService,
    8.             PartialDispatcherServices partialDispatcherServices)
    9.             throws Exception {
    10.  
    11.         final DispatcherLeaderProcessFactory dispatcherLeaderProcessFactory =
    12.                 dispatcherLeaderProcessFactoryFactory.createFactory(
    13.                         jobGraphStoreFactory,
    14.                         ioExecutor,
    15.                         rpcService,
    16.                         partialDispatcherServices,
    17.                         fatalErrorHandler);
    18.  
    19.         // TODO
    20.         return DefaultDispatcherRunner.create(
    21.                 leaderElectionService, fatalErrorHandler, dispatcherLeaderProcessFactory);
    22.     }

    根据变量名，我们可以看出在这里构建了一个Dispatcher的Leader竞选线程工厂，并将该对象作为参数传入了DispatcherRunner的构建方法里，我们进入DefaultDispatcherRunner.create方法：

    1.     public static DispatcherRunner create(
    2.             LeaderElectionService leaderElectionService,
    3.             FatalErrorHandler fatalErrorHandler,
    4.             DispatcherLeaderProcessFactory dispatcherLeaderProcessFactory)
    5.             throws Exception {
    6.         final DefaultDispatcherRunner dispatcherRunner =
    7.                 new DefaultDispatcherRunner(
    8.                         leaderElectionService, fatalErrorHandler, dispatcherLeaderProcessFactory);
    9.         // TODO 进入此方法
    10.         return DispatcherRunnerLeaderElectionLifecycleManager.createFor(
    11.                 dispatcherRunner, leaderElectionService);
    12.     }

    在构建了DispatcherRunner之后，将该实例传入了DispatcherRunner竞选Leader的生命周期管理方法，我们进入DispatcherRunnerLeaderElectionLifecycleManager.createFor方法继续分析

    1.     public static extends DispatcherRunner & LeaderContender> DispatcherRunner createFor(
    2.             T dispatcherRunner, LeaderElectionService leaderElectionService) throws Exception {
    3.         // TODO 来看构造方法
    4.         return new DispatcherRunnerLeaderElectionLifecycleManager<>(
    5.                 dispatcherRunner, leaderElectionService);
    6.     }

    继续进入DispatcherRunnerLeaderElectionLifecycleManager的构造方法：

    1.     private DispatcherRunnerLeaderElectionLifecycleManager(
    2.             T dispatcherRunner, LeaderElectionService leaderElectionService) throws Exception {
    3.         this.dispatcherRunner = dispatcherRunner;
    4.         this.leaderElectionService = leaderElectionService;
    5.  
    6.         // TODO 开始竞选,竞选者为 dispatcherRunner
    7.         leaderElectionService.start(dispatcherRunner);
    8.     }

    又看到了我们熟悉的方法，开始Leader竞选！

    2.1、DispatcherRunner的Leader选举

    我们进入start方法，选择DefaultLeaderElectionService实现：

    1.     @Override
    2.     public final void start(LeaderContender contender) throws Exception {
    3.         checkNotNull(contender, "Contender must not be null.");
    4.         Preconditions.checkState(leaderContender == null, "Contender was already set.");
    5.  
    6.         synchronized (lock) {
    7.             /*
    8.              TODO 在WebMonitorEndpoint中调用时，此contender为DispatcherRestEndPoint
    9.               在ResourceManager中调用时,contender为ResourceManager
    10.               在DispatcherRunner中调用时,contender为DispatcherRunner
    11.              */
    12.             leaderContender = contender;
    13.  
    14.             // TODO 此处创建选举对象 leaderElectionDriver
    15.             leaderElectionDriver =
    16.                     leaderElectionDriverFactory.createLeaderElectionDriver(
    17.                             this,
    18.                             new LeaderElectionFatalErrorHandler(),
    19.                             leaderContender.getDescription());
    20.             LOG.info("Starting DefaultLeaderElectionService with {}.", leaderElectionDriver);
    21.  
    22.             running = true;
    23.         }
    24.     }

    又是熟悉的方法，在前两章中，ResourceManager、WebMonitorEndpoint组件的Leader竞选都使用的该方法，此处是DispatcherRunner的竞选，所以此处的contender为DispatcherRunner，我们继续看竞选流程，进入leaderElectionDriverFactory.createLeaderElectionDriver方法，由于是基于standalone模式分析源码，Leader的竞选依赖于zookeeper，我们进入ZooKeeperLeaderElectionDriverFactory实现：

    1.     @Override
    2.     public ZooKeeperLeaderElectionDriver createLeaderElectionDriver(
    3.             LeaderElectionEventHandler leaderEventHandler,
    4.             FatalErrorHandler fatalErrorHandler,
    5.             String leaderContenderDescription)
    6.             throws Exception {
    7.         return new ZooKeeperLeaderElectionDriver(
    8.                 client,
    9.                 latchPath,
    10.                 leaderPath,
    11.                 leaderEventHandler,
    12.                 fatalErrorHandler,
    13.                 leaderContenderDescription);
    14.     }

    再进入ZooKeeperLeaderElectionDriver的构造方法：

    1.     public ZooKeeperLeaderElectionDriver(
    2.             CuratorFramework client,
    3.             String latchPath,
    4.             String leaderPath,
    5.             LeaderElectionEventHandler leaderElectionEventHandler,
    6.             FatalErrorHandler fatalErrorHandler,
    7.             String leaderContenderDescription)
    8.             throws Exception {
    9.         this.client = checkNotNull(client);
    10.         this.leaderPath = checkNotNull(leaderPath);
    11.         this.leaderElectionEventHandler = checkNotNull(leaderElectionEventHandler);
    12.         this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
    13.         this.leaderContenderDescription = checkNotNull(leaderContenderDescription);
    14.  
    15.         leaderLatch = new LeaderLatch(client, checkNotNull(latchPath));
    16.         cache = new NodeCache(client, leaderPath);
    17.  
    18.         client.getUnhandledErrorListenable().addListener(this);
    19.  
    20.         running = true;
    21.  
    22.         // TODO 开始选举
    23.         leaderLatch.addListener(this);
    24.         leaderLatch.start();
    25.  
    26.         /*
    27.         TODO 选举开始后，不就会接收到响应：
    28.          1.如果竞选成功，则回调该类的isLeader方法
    29.          2.如果竞选失败，则回调该类的notLeader方法
    30.          每一个竞选者对应一个竞选Driver
    31.          */
    32.  
    33.         cache.getListenable().addListener(this);
    34.         cache.start();
    35.  
    36.         client.getConnectionStateListenable().addListener(listener);
    37.     }

    又是熟悉的地方，根据前两章的分析，Leader竞选完成后会根据竞选结果回调isLeader方法或notLeader方法，此处我们直接去看isLeader方法：

    1.  /*
    2.     选举成功
    3.      */
    4.     @Override
    5.     public void isLeader() {
    6.         leaderElectionEventHandler.onGrantLeadership();
    7.     }

    在点进来：

    1.  @Override
    2.     @GuardedBy("lock")
    3.     public void onGrantLeadership() {
    4.         synchronized (lock) {
    5.             if (running) {
    6.                 issuedLeaderSessionID = UUID.randomUUID();
    7.                 clearConfirmedLeaderInformation();
    8.  
    9.                 if (LOG.isDebugEnabled()) {
    10.                     LOG.debug(
    11.                             "Grant leadership to contender {} with session ID {}.",
    12.                             leaderContender.getDescription(),
    13.                             issuedLeaderSessionID);
    14.                 }
    15.  
    16.                 /*
    17.                 TODO 有4中竞选者类型，LeaderContender有4中情况
    18.                  1.Dispatcher = DefaultDispatcherRunner
    19.                  2.JobMaster = JobManagerRunnerImpl
    20.                  3.ResourceManager = ResourceManager
    21.                  4.WebMonitorEndpoint = WebMonitorEndpoint
    22.                  */
    23.                 leaderContender.grantLeadership(issuedLeaderSessionID);
    24.             } else {
    25.                 if (LOG.isDebugEnabled()) {
    26.                     LOG.debug(
    27.                             "Ignoring the grant leadership notification since the {} has "
    28.                                     + "already been closed.",
    29.                             leaderElectionDriver);
    30.                 }
    31.             }
    32.         }
    33.     }

    再进入leaderContender.grantLeadership方法，由于当前是DispatcherRunner的选举，我们选择DefaultDispatcherRunner实现：

    1.  // ---------------------------------------------------------------
    2.     // Leader election
    3.     // ---------------------------------------------------------------
    4.  
    5.     @Override
    6.     public void grantLeadership(UUID leaderSessionID) {
    7.         runActionIfRunning(
    8.                 () -> {
    9.                     LOG.info(
    10.                             "{} was granted leadership with leader id {}. Creating new {}.",
    11.                             getClass().getSimpleName(),
    12.                             leaderSessionID,
    13.                             DispatcherLeaderProcess.class.getSimpleName());
    14.                     // TODO
    15.                     startNewDispatcherLeaderProcess(leaderSessionID);
    16.                 });
    17.     }

    根据方法名不难猜出，接下来是启动一个新的DispatcherLeader，我们进入startNewDispatcherLeaderProcess方法：

    1.     private void startNewDispatcherLeaderProcess(UUID leaderSessionID) {
    2.         // TODO 如果当前有DispatcherLeader则先关闭
    3.         stopDispatcherLeaderProcess();
    4.  
    5.         // TODO 然后再创建
    6.         dispatcherLeaderProcess = createNewDispatcherLeaderProcess(leaderSessionID);
    7.  
    8.         final DispatcherLeaderProcess newDispatcherLeaderProcess = dispatcherLeaderProcess;
    9.         FutureUtils.assertNoException(
    10.                 previousDispatcherLeaderProcessTerminationFuture.thenRun(
    11.                         // TODO 启动
    12.                         newDispatcherLeaderProcess::start));
    13.     }

    在该方法里一共做了三件事：

    1、先判断当前是否有正在运行的DispatcherLeader，如果有则先关闭，保证当前环境中只有一个且是最新的DispatcherLeader。

    2、然后再创建DispatcherLeader

    3、启动DispatcherLeader

    2.2、开始准备构建Dispatcher

    我们来看newDispatcherLeaderProcess的start方法：

    1.   @Override
    2.     public final void start() {
    3.         // TODO
    4.         runIfStateIs(State.CREATED, this::startInternal);
    5.     }
    6.  
    7.     private void startInternal() {
    8.         log.info("Start {}.", getClass().getSimpleName());
    9.         state = State.RUNNING;
    10.         // TODO
    11.         onStart();
    12.     }

    再来看startInternal的onStart方法，选择SessionDispatcherLeaderProcess实现：

    1.     @Override
    2.     protected void onStart() {
    3.         // TODO 启动Dispatcher服务,启动JobGraphStore
    4.         startServices();
    5.  
    6.         // TODO 异步编程, 若JobGraphStore启动后发现内部有未执行完毕的Job,则先通过recoverJobsAsync恢复JobGraph
    7.         // TODO 再用过createDispatcherIfRunning启动Dispatcher
    8.         onGoingRecoveryOperation =
    9.                 recoverJobsAsync()
    10.                         // TODO 构建Dispatcher并启动
    11.                         .thenAccept(this::createDispatcherIfRunning)
    12.                         .handle(this::onErrorIfRunning);
    13.     }
    14.

    在这个方法里一共做了三件事：

    1、启动Dispatcher所需的基础服务，启动JobGraphStore

    2、恢复之前因为非正常原因没有执行完的Job

    3、构建并启动Dispatcher

    下面我们来详细聊聊这几个部分

    2.2.1、启动JobGraphStore

    我们先来看JobGraphStore的启动，进入startServices方法：

    1.     private void startServices() {
    2.         try {
    3.             // TODO 启动JobGraphStore
    4.             jobGraphStore.start(this);
    5.         } catch (Exception e) {
    6.             throw new FlinkRuntimeException(
    7.                     String.format(
    8.                             "Could not start %s when trying to start the %s.",
    9.                             jobGraphStore.getClass().getSimpleName(), getClass().getSimpleName()),
    10.                     e);
    11.         }
    12.     }

    进入start方法，选择DefaultJobGraphStore实现：

    1.     @Override
    2.     public void start(JobGraphListener jobGraphListener) throws Exception {
    3.         synchronized (lock) {
    4.             if (!running) {
    5.                 // TODO 启动监听
    6.                 // TODO 此处的监听,若有JobGraph添加则会回调 onAddedJobGraph方法
    7.                 // TODO 若有JobGraph删除则会回调 onRemovedJobGraph 方法
    8.                 this.jobGraphListener = checkNotNull(jobGraphListener);
    9.                 jobGraphStoreWatcher.start(this);
    10.                 running = true;
    11.             }
    12.         }
    13.     }

    可以看到此处启动了一个JobGraph的监听服务，当有JobGraph提交进来时会触发onAddedJobGraph方法，当有JobGraph移除时会回调onRemovedJobGraph方法，详细内容我们会在后续的Job提交源码分析力介绍。现在我们回到之前的onStart方法

    2.2.2、寻找中断的Job

            若JobGraphStore启动后发现内部有未执行完毕的Job，在recoverJobsAsync()方法里会遍历这些Job并加入集合中：

    1.     private Collection recoverJobs() {
    2.         log.info("Recover all persisted job graphs.");
    3.         final Collection jobIds = getJobIds();
    4.         final Collection recoveredJobGraphs = new ArrayList<>();
    5.  
    6.         for (JobID jobId : jobIds) {
    7.             recoveredJobGraphs.add(recoverJob(jobId));
    8.         }
    9.  
    10.         log.info("Successfully recovered {} persisted job graphs.", recoveredJobGraphs.size());
    11.  
    12.         return recoveredJobGraphs;
    13.     }

    2.2.3、构建Dispatcher并启动

    在完成中断Job的恢复工作后，开始真正的构建Dispatcher实例，并启动，我们来看createDispatcherIfRunning方法：

    1.     private void createDispatcherIfRunning(Collection jobGraphs) {
    2.         runIfStateIs(State.RUNNING, () -> createDispatcher(jobGraphs));
    3.     }

    再进入createDispatcher方法：

    1.     private void createDispatcher(Collection jobGraphs) {
    2.  
    3.         final DispatcherGatewayService dispatcherService =
    4.                 // TODO 构建Dispatcher并启动
    5.                 dispatcherGatewayServiceFactory.create(
    6.                         DispatcherId.fromUuid(getLeaderSessionId()), jobGraphs, jobGraphStore);
    7.  
    8.         completeDispatcherSetup(dispatcherService);
    9.     }

    可以看到此处已经开始构建Dispatcher了，我们再点入create方法，选择DefaultDispatcherGatewayServiceFactory实现：

    1.     @Override
    2.     public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
    3.             DispatcherId fencingToken,
    4.             Collection recoveredJobs,
    5.             JobGraphWriter jobGraphWriter) {
    6.  
    7.         final Dispatcher dispatcher;
    8.         try {
    9.             // TODO 构建Dispatcher
    10.             dispatcher =
    11.                     dispatcherFactory.createDispatcher(
    12.                             rpcService,
    13.                             fencingToken,
    14.                             recoveredJobs,
    15.                             (dispatcherGateway, scheduledExecutor, errorHandler) ->
    16.                                     new NoOpDispatcherBootstrap(),
    17.                             PartialDispatcherServicesWithJobGraphStore.from(
    18.                                     partialDispatcherServices, jobGraphWriter));
    19.         } catch (Exception e) {
    20.             throw new FlinkRuntimeException("Could not create the Dispatcher rpc endpoint.", e);
    21.         }
    22.  
    23.         // TODO 启动DIspatcher
    24.         dispatcher.start();
    25.  
    26.         return DefaultDispatcherGatewayService.from(dispatcher);
    27.     }

    可以看到在这里真正构建了Dispatcher实例，并调用了start方法启动Dispatcher，我们先来看createDispatcher方法，选择SessionDispatcherFactory实现：

    1.     @Override
    2.     public StandaloneDispatcher createDispatcher(
    3.             RpcService rpcService,
    4.             DispatcherId fencingToken,
    5.             Collection recoveredJobs,
    6.             DispatcherBootstrapFactory dispatcherBootstrapFactory,
    7.             PartialDispatcherServicesWithJobGraphStore partialDispatcherServicesWithJobGraphStore)
    8.             throws Exception {
    9.         // create the default dispatcher
    10.         // TODO 继承了RpcEndpoint,创建完成后会回调onStart方法
    11.         return new StandaloneDispatcher(
    12.                 rpcService,
    13.                 fencingToken,
    14.                 recoveredJobs,
    15.                 dispatcherBootstrapFactory,
    16.                 DispatcherServices.from(
    17.                         partialDispatcherServicesWithJobGraphStore,
    18.                         JobMasterServiceLeadershipRunnerFactory.INSTANCE));
    19.     }

    我们再来看StandaloneDispatcher的构造方法：

    1. public class StandaloneDispatcher extends Dispatcher {
    2.     public StandaloneDispatcher(
    3.             RpcService rpcService,
    4.             DispatcherId fencingToken,
    5.             Collection recoveredJobs,
    6.             DispatcherBootstrapFactory dispatcherBootstrapFactory,
    7.             DispatcherServices dispatcherServices)
    8.             throws Exception {
    9.         super(
    10.                 rpcService,
    11.                 fencingToken,
    12.                 recoveredJobs,
    13.                 dispatcherBootstrapFactory,
    14.                 dispatcherServices);
    15.     }
    16. }

    再进入super，我们来到了Dispatcher类内部，因为Dispatcher继承了RpcEndpoint，根据我们在FlinkRPC章节讲到的内容，此刻我们知道在Dispatcher初始化之后会调用onStart方法，我们直接去看onStart方法：

    1. // ------------------------------------------------------
    2.     // Lifecycle methods
    3.     // ------------------------------------------------------
    4.  
    5.     @Override
    6.     public void onStart() throws Exception {
    7.         try {
    8.             // TODO 启动Dispatcher基础服务
    9.             startDispatcherServices();
    10.         } catch (Throwable t) {
    11.             final DispatcherException exception =
    12.                     new DispatcherException(
    13.                             String.format("Could not start the Dispatcher %s", getAddress()), t);
    14.             onFatalError(exception);
    15.             throw exception;
    16.         }
    17.  
    18.         // TODO 启动待恢复的Job
    19.         startRecoveredJobs();
    20.         this.dispatcherBootstrap =
    21.                 this.dispatcherBootstrapFactory.create(
    22.                         getSelfGateway(DispatcherGateway.class),
    23.                         this.getRpcService().getScheduledExecutor(),
    24.                         this::onFatalError);
    25.     }

    这里做了三件事：

    1、启动Dispatcher的基础服务

    2、开始恢复之前添加到集合中的中断的Job

    3、构建DIspatcher实例

    在Dispatcher的基础服务中只启动了一个Metric服务，没什么好看的，我们来看中断Job的恢复：

    1.     private void startRecoveredJobs() {
    2.         for (JobGraph recoveredJob : recoveredJobs) {
    3.             runRecoveredJob(recoveredJob);
    4.         }
    5.         recoveredJobs.clear();
    6.     }
    7.  
    8.     private void runRecoveredJob(final JobGraph recoveredJob) {
    9.         checkNotNull(recoveredJob);
    10.         try {
    11.             // TODO 以Recover模式运行Job
    12.             // TODO 内部具体实现等后面分析作业提交流程时再来分析
    13.             runJob(recoveredJob, ExecutionType.RECOVERY);
    14.         } catch (Throwable throwable) {
    15.             onFatalError(
    16.                     new DispatcherException(
    17.                             String.format(
    18.                                     "Could not start recovered job %s.", recoveredJob.getJobID()),
    19.                             throwable));
    20.         }
    21.     }

    我们可以看到，此处会遍历之前的中断Job集合，并对每一个中断Job以RECOVER模式恢复运行，具体的实现我们后面再来分析。我们继续来看Dispatcher的构建，回到之前的方法，我们俩看dispatcherBootstrapFactory.create，选择DefaultDispatcherGatewayServiceFactory，我们又回到了这里：

    1. @Override
    2.     public AbstractDispatcherLeaderProcess.DispatcherGatewayService create(
    3.             DispatcherId fencingToken,
    4.             Collection recoveredJobs,
    5.             JobGraphWriter jobGraphWriter) {
    6.  
    7.         final Dispatcher dispatcher;
    8.         try {
    9.             // TODO 构建Dispatcher
    10.             dispatcher =
    11.                     dispatcherFactory.createDispatcher(
    12.                             rpcService,
    13.                             fencingToken,
    14.                             recoveredJobs,
    15.                             (dispatcherGateway, scheduledExecutor, errorHandler) ->
    16.                                     new NoOpDispatcherBootstrap(),
    17.                             PartialDispatcherServicesWithJobGraphStore.from(
    18.                                     partialDispatcherServices, jobGraphWriter));
    19.         } catch (Exception e) {
    20.             throw new FlinkRuntimeException("Could not create the Dispatcher rpc endpoint.", e);
    21.         }
    22.  
    23.         // TODO 启动DIspatcher
    24.         dispatcher.start();
    25.  
    26.         return DefaultDispatcherGatewayService.from(dispatcher);
    27.     }

    至此，Dispatcher实例已经构建完毕，接下来就是启动Dispatcher，在start方法里，Dispatcher向自己发送了一条消息，告知已启动完毕：

    1.     @Override
    2.     public void start() {
    3.         // 向自己发送消息,告知已启动
    4.         rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
    5.     }

    到此为止，Dispatcher服务已构建完毕也已启动完毕，我们总结一下。

    三、总结

    Dispatcher的构建其实一共就做了两件事：

    1、启动 JobGraphStore 服务

    2、从 JobGraphStrore 恢复执行 Job, 要启动 Dispatcher

    只不过Dispatcher的构建之前，Flink先构建了一个DispatcherRunner，并进行了Leader选举，选举完成之后才由LeaderDispatcherRunner构建Dispatcher并启动。在这里需要注意两点：

    1、DispatcherRunner的选举环节会回调isLeader方法。

    2、Dispatcher对象继承了RpcEndpoint，所以在构建完成后会调用onStart方法。

    在前三章中，我们介绍了主节点（逻辑JobManager）的启动流程，以及8大基础服务的构建和启动，并且在前两章中我们介绍了WebMonitorEndpoint组件和ResourceManager组价你的启动，到此为止Dispatcher也已启动完毕，主节点也在这里完成了它所有的启动工作。在下一章中，我们来看看从节点TaskManager的启动流程！

  • 相关阅读:
    Java基础进阶-序列化
    python高校闲置物品交换系统django
    HX_JavaSE_day01
    java性能测试
    【遥控器开发基础教程5】疯壳·开源编队无人机-SPI(2.4G 双机通信)
    C#使用词嵌入向量与向量数据库为大语言模型(LLM)赋能长期记忆实现私域问答机器人落地
    ubuntu为可执行程序添加桌面图标
    数据结构和算法——睡眠排序
    【Oracle】Oracle清理日志空间
    芯片SoC设计你了解吗？
  • 原文地址：https://blog.csdn.net/EdwardWong_/article/details/126555250