flink源码分析之TaskManager启动篇

    if (useSingleRpcService) {
                    // we always need the 'commonRpcService' for auxiliary calls
                    // 这里会创建local rpc service
                    commonRpcService = createLocalRpcService(configuration);
                    final CommonRpcServiceFactory commonRpcServiceFactory = new CommonRpcServiceFactory(commonRpcService);
                    taskManagerRpcServiceFactory = commonRpcServiceFactory;
                    dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;
                    metricQueryServiceRpcService = MetricUtils.startLocalMetricsRpcService(configuration);
    } else {
        ------
            dispatcherResourceManagreComponentRpcServiceFactory =
            new DedicatedRpcServiceFactory(
            configuration,
            // jobManager地址
            jobManagerExternalAddress,
            jobManagerExternalPortRange,
            jobManagerBindAddress);
        taskManagerRpcServiceFactory =
            new DedicatedRpcServiceFactory(
            configuration,
            // taskManager地址
            taskManagerExternalAddress,
            taskManagerExternalPortRange,
            taskManagerBindAddress);
    }   
 --------------省略部分代码----------------       
// 启动QueryService,在MetricRegistryImpl中，会启动一个rpc server
metricRegistry.startQueryService(metricQueryServiceRpcService, null);
// 执行io操作的线程池
ioExecutor = Executors.newFixedThreadPool(
ClusterEntrypointUtils.getPoolSize(configuration),
new ExecutorThreadFactory("mini-cluster-io"));
// 高可用服务
haServices = createHighAvailabilityServices(configuration, ioExecutor);
// blobServer用于管理一些jar包等
blobServer = new BlobServer(configuration, haServices.createBlobStore());
blobServer.start();
// 心跳服务
heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
// blob缓存服务，用于BroadCast的缓存管理等
blobCacheService = new BlobCacheService(
configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())
);
// 启动taskManager，会启动taskExecutor，也会启动一个rpcServer
startTaskManagers();

protected RpcService createLocalRpcService(Configuration configuration) throws Exception {
        return AkkaRpcServiceUtils.localServiceBuilder(configuration)
            .withCustomConfig(AkkaUtils.testDispatcherConfig())
            .createAndStart();
    }

        @Override
        public RpcService createRpcService() throws Exception {
            final RpcService rpcService = MiniCluster.this.createRemoteRpcService(
                configuration, externalAddress, externalPortRange, bindAddress);
            synchronized (lock) {
                rpcServices.add(rpcService);
            }
            return rpcService;
        }
        protected RpcService createRemoteRpcService(
            Configuration configuration,
            String externalAddress,
            String externalPortRange,
            String bindAddress) throws Exception {
            return AkkaRpcServiceUtils.remoteServiceBuilder(configuration, externalAddress, externalPortRange)
                // 需要传入绑定地址
                .withBindAddress(bindAddress)
                .withCustomConfig(AkkaUtils.testDispatcherConfig())
                .createAndStart();
    }

public AkkaRpcService createAndStart() throws Exception {
        ------------------省略部分---------------
            final ActorSystem actorSystem;
            if (externalAddress == null) {
                // create local actor system
                // 创建本地的actor system，不需要外部地址，所有服务共用同一套actorSystem
                actorSystem = BootstrapTools.startLocalActorSystem(
                    configuration,
                    actorSystemName,
                    logger,
                    actorSystemExecutorConfiguration,
                    customConfig);
            } else {
                // create remote actor system
                // 因为是与外部地址绑定的，所以对于jobManager和TaskManager来说各自有一套actkorSystem
                actorSystem = BootstrapTools.startRemoteActorSystem(
                    configuration,
                    actorSystemName,
                    externalAddress,
                    externalPortRange,
                    bindAddress,
                    Optional.ofNullable(bindPort),
                    logger,
                    actorSystemExecutorConfiguration,
                    customConfig);
            }
            return new AkkaRpcService(actorSystem, AkkaRpcServiceConfiguration.fromConfiguration(configuration));
        }

@VisibleForTesting
    public AkkaRpcService(final ActorSystem actorSystem, final AkkaRpcServiceConfiguration configuration) {
        this.actorSystem = checkNotNull(actorSystem, "actor system");
        this.configuration = checkNotNull(configuration, "akka rpc service configuration");
        // 需要传入一个actorSystem
        Address actorSystemAddress = AkkaUtils.getAddress(actorSystem);
        if (actorSystemAddress.host().isDefined()) {
            address = actorSystemAddress.host().get();
        } else {
            address = "";
        }
        if (actorSystemAddress.port().isDefined()) {
            port = (Integer) actorSystemAddress.port().get();
        } else {
            port = -1;
        }
        captureAskCallstacks = configuration.captureAskCallStack();
        // 内部调度执行器
        internalScheduledExecutor = new ActorSystemScheduledExecutorAdapter(actorSystem);
        terminationFuture = new CompletableFuture<>();
        stopped = false;
        // actor体系中的监督者，主要用于服务重启，对于flink的重启机制发挥作用的地方，它有自己的executorService
        supervisor = startSupervisorActor();
    }

    protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
        this.rpcService = checkNotNull(rpcService, "rpcService");
        this.endpointId = checkNotNull(endpointId, "endpointId");
        this.rpcServer = rpcService.startServer(this);
        this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
    }

    @Override
    public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
        checkNotNull(rpcEndpoint, "rpc endpoint");
        // 注册akkaRpcActor，TaskExecutor不是FencedRpcEndpoint类型的，JobMaster是FencedRpcEndpoint类型的
        final SupervisorActor.ActorRegistration actorRegistration = registerAkkaRpcActor(rpcEndpoint);
        // 获取注册后得到的actor对象
        final ActorRef actorRef = actorRegistration.getActorRef();
        // TerminationFuture，可以在里面执行一些中止后的操作
        final CompletableFuture<Void> actorTerminationFuture = actorRegistration.getTerminationFuture();
        LOG.info("Starting RPC endpoint for {} at {} .", rpcEndpoint.getClass().getName(), actorRef.path());
        // 获得对应actor对象在对应的actorSystem中的地址
        final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);
        final String hostname;
        Option<String> host = actorRef.path().address().host();
        if (host.isEmpty()) {
            hostname = "localhost";
        } else {
            hostname = host.get();
        }
        Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));
        // 为代理接口中添加RpcServer接口
        implementedRpcGateways.add(RpcServer.class);
        // 添加AkkaBasedEndpoint接口
        implementedRpcGateways.add(AkkaBasedEndpoint.class);
        // 用于触发代理类方法的invocationHandler
        final InvocationHandler akkaInvocationHandler;
        // 如果是FencedRpcEndpoint，这里对应的akkaInvocationHandler为FencedAkkaInvocationHandler类型，JobMaster返回的是这种类型的
        if (rpcEndpoint instanceof FencedRpcEndpoint) {
            // a FencedRpcEndpoint needs a FencedAkkaInvocationHandler
            akkaInvocationHandler = new FencedAkkaInvocationHandler<>(
                akkaAddress,
                hostname,
                actorRef,
                configuration.getTimeout(),
                configuration.getMaximumFramesize(),
                actorTerminationFuture,
                ((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken,
                captureAskCallstacks);
           implementedRpcGateways.add(FencedMainThreadExecutable.class);
        } else {// 如果不是FencedRpcEndpoint，这里返回的是AkkaInvocationHandler，TaskExecutor返回的是这种类型的
            akkaInvocationHandler = new AkkaInvocationHandler(
                akkaAddress,
                hostname,
                actorRef,
                configuration.getTimeout(),
                configuration.getMaximumFramesize(),
                actorTerminationFuture,
                captureAskCallstacks);
        }
        // Rather than using the System ClassLoader directly, we derive the ClassLoader
        // from this class . That works better in cases where Flink runs embedded and all Flink
        // code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader
        ClassLoader classLoader = getClass().getClassLoader();
        // jdk的动态代理，对implementedRpcGateways里维护的接口实现进行代理，会在akkaInvocationHandler的invoke方法中反射调用接口中的一些方法
        @SuppressWarnings("unchecked")
        RpcServer server = (RpcServer) Proxy.newProxyInstance(
            classLoader,
            implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]),
            akkaInvocationHandler);
        return server;
    }

@Override
    public void start() {
        rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
    }

@GuardedBy("lock")
private void startTaskManagers() throws Exception {
    final int numTaskManagers = miniClusterConfiguration.getNumTaskManagers();
    LOG.info("Starting {} TaskManger(s)", numTaskManagers);
    for (int i = 0; i < numTaskManagers; i++) {
        startTaskExecutor();
    }
}

@VisibleForTesting
    void startTaskExecutor() throws Exception {
        synchronized (lock) {
            final Configuration configuration = miniClusterConfiguration.getConfiguration();
            // 创建TaskExecutor
            final TaskExecutor taskExecutor = TaskManagerRunner.startTaskManager(
                configuration,
                // 这个用于在ResourceManager中标识TaskExecutor的使用资源
                new ResourceID(UUID.randomUUID().toString()),
                taskManagerRpcServiceFactory.createRpcService(),
                haServices,
                heartbeatServices,
                metricRegistry,
                blobCacheService,
                useLocalCommunication(),
                ExternalResourceInfoProvider.NO_EXTERNAL_RESOURCES,
                taskManagerTerminatingFatalErrorHandlerFactory.create(taskManagers.size()));
            // 启动TaskExecutor
            taskExecutor.start();
            // 将taskExecutor加入到taskMangers列表中
            taskManagers.add(taskExecutor);
        }
    }

public static TaskExecutor startTaskManager(
            Configuration configuration,
            ResourceID resourceID,
            RpcService rpcService,
            HighAvailabilityServices highAvailabilityServices,
            HeartbeatServices heartbeatServices,
            MetricRegistry metricRegistry,
            BlobCacheService blobCacheService,
            boolean localCommunicationOnly,
            ExternalResourceInfoProvider externalResourceInfoProvider,
            FatalErrorHandler fatalErrorHandler) throws Exception {
   -----------------省略部分代码------------------------------------
        String externalAddress = rpcService.getAddress();
        // 声明运行TaskExecutor所需要的资源
        final TaskExecutorResourceSpec taskExecutorResourceSpec = TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);
        // TaskManager服务配置
        TaskManagerServicesConfiguration taskManagerServicesConfiguration =
            TaskManagerServicesConfiguration.fromConfiguration(
                configuration,
                resourceID,
                externalAddress,
                localCommunicationOnly,
                taskExecutorResourceSpec);
        // taskManager的Metric组，用于metric管理
        Tuple2<TaskManagerMetricGroup, MetricGroup> taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
            metricRegistry,
            externalAddress,
            resourceID,
            taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
        // taskExecutor的执行线程池，它的个数是根据cluster.io-pool.size参数配置，如果没有配置会使用cpu核数的四倍
        final ExecutorService ioExecutor = Executors.newFixedThreadPool(
            taskManagerServicesConfiguration.getNumIoThreads(),
            new ExecutorThreadFactory("flink-taskexecutor-io"));
        // TaskManager的一系列服务
        TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
            taskManagerServicesConfiguration,
            blobCacheService.getPermanentBlobService(),
            taskManagerMetricGroup.f1,
            ioExecutor,
            fatalErrorHandler);
             ------------------省略部分代码---------------------
        return new TaskExecutor(
            rpcService,
            taskManagerConfiguration,
            highAvailabilityServices,
            taskManagerServices,
            externalResourceInfoProvider,
            heartbeatServices,
            taskManagerMetricGroup.f0,
            metricQueryServiceAddress,
            blobCacheService,
            fatalErrorHandler,
            new TaskExecutorPartitionTrackerImpl(taskManagerServices.getShuffleEnvironment()),
            createBackPressureSampleService(configuration, rpcService.getScheduledExecutor()));         
    }

public static TaskManagerServices fromConfiguration(
            TaskManagerServicesConfiguration taskManagerServicesConfiguration,
            PermanentBlobService permanentBlobService,
            MetricGroup taskManagerMetricGroup,
            ExecutorService ioExecutor,
            FatalErrorHandler fatalErrorHandler) throws Exception {
        // pre-start checks
        checkTempDirs(taskManagerServicesConfiguration.getTmpDirPaths());
        // 类似于一个even source模型，内部维护着一个分区与TaskEventHandler之间关系的map，其他线程可以
        // 执行subscribe进行事件订阅，在有事件发生时TaskEventDispatcher会执行publish方法进行通知，主要为了达到某个地方发生改变后，关注的位置也相应作出反应。
        final TaskEventDispatcher taskEventDispatcher = new TaskEventDispatcher();
        // start the I/O manager, it will create some temp directories.
        // 异步的io管理器，主要用于控制io操作
        final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
         // shuffle环境，在这个方法内部会有一个SPI的拓展点，会从配置的shuffle-service-factory.class中加载用户定义的shuffle-service-factory。内部提供了一个NettyShuffleServiceFactory的实现，会创建NettyShuffleEnvironment对象
        final ShuffleEnvironment<?, ?> shuffleEnvironment = createShuffleEnvironment(
            taskManagerServicesConfiguration,
            taskEventDispatcher,
            taskManagerMetricGroup,
            ioExecutor);
        // 启动shuffleEnvironment，如果使用的是NettyShuffleServiceFactory这里会启动一个netty server和netty client，并和dubbo 协议类似用一个ConnectionId维护一个PartitionRequestClient,关于这个内部的细节后续会详细来分析
        final int listeningDataPort = shuffleEnvironment.start();
        // 创建kvState服务
        final KvStateService kvStateService = KvStateService.fromConfiguration(taskManagerServicesConfiguration);
           // 启动kvState服务
        kvStateService.start();
        // 未能解析的taskManager的位置
        final UnresolvedTaskManagerLocation unresolvedTaskManagerLocation = new UnresolvedTaskManagerLocation(
            taskManagerServicesConfiguration.getResourceID(),
            taskManagerServicesConfiguration.getExternalAddress(),
            // we expose the task manager location with the listening port
            // iff the external data port is not explicitly defined
            taskManagerServicesConfiguration.getExternalDataPort() > 0 ?
                taskManagerServicesConfiguration.getExternalDataPort() :
                listeningDataPort);
        // 广播变量管理器
        final BroadcastVariableManager broadcastVariableManager = new BroadcastVariableManager();
        // 任务槽table
        final TaskSlotTable<Task> taskSlotTable = createTaskSlotTable(
            taskManagerServicesConfiguration.getNumberOfSlots(),
            taskManagerServicesConfiguration.getTaskExecutorResourceSpec(),
            taskManagerServicesConfiguration.getTimerServiceShutdownTimeout(),
            taskManagerServicesConfiguration.getPageSize(),
            ioExecutor);
        // 任务table
        final JobTable jobTable = DefaultJobTable.create();
        // 任务leader服务
        final JobLeaderService jobLeaderService = new DefaultJobLeaderService(unresolvedTaskManagerLocation, taskManagerServicesConfiguration.getRetryingRegistrationConfiguration());
        // state根目录
        final String[] stateRootDirectoryStrings = taskManagerServicesConfiguration.getLocalRecoveryStateRootDirectories();
        final File[] stateRootDirectoryFiles = new File[stateRootDirectoryStrings.length];
        for (int i = 0; i < stateRootDirectoryStrings.length; ++i) {
            stateRootDirectoryFiles[i] = new File(stateRootDirectoryStrings[i], LOCAL_STATE_SUB_DIRECTORY_ROOT);
        }
        // taskState管理器
        final TaskExecutorLocalStateStoresManager taskStateManager = new TaskExecutorLocalStateStoresManager(
            taskManagerServicesConfiguration.isLocalRecoveryEnabled(),
            stateRootDirectoryFiles,
            ioExecutor);
         // 当jvm的元数据空间内存溢出时的错误
        final boolean failOnJvmMetaspaceOomError =
taskManagerServicesConfiguration.getConfiguration().getBoolean(CoreOptions.FAIL_ON_USER_CLASS_LOADING_METASPACE_OOM);
        // 库缓存管理器
        final LibraryCacheManager libraryCacheManager = new BlobLibraryCacheManager(
            permanentBlobService,
            BlobLibraryCacheManager.defaultClassLoaderFactory(
                taskManagerServicesConfiguration.getClassLoaderResolveOrder(),
                taskManagerServicesConfiguration.getAlwaysParentFirstLoaderPatterns(),
                failOnJvmMetaspaceOomError ? fatalErrorHandler : null));
        // 返回TaskManager服务
        return new TaskManagerServices(
            unresolvedTaskManagerLocation,
            taskManagerServicesConfiguration.getManagedMemorySize().getBytes(),
            ioManager,
            shuffleEnvironment,
            kvStateService,
            broadcastVariableManager,
            taskSlotTable,
            jobTable,
            jobLeaderService,
            taskStateManager,
            taskEventDispatcher,
            ioExecutor,
            libraryCacheManager);
    }

  private final Map<JobID, JobOrConnection> jobs;
      private final Map<ResourceID, JobID> resourceIdJobIdIndex;
      private DefaultJobTable() {
          this.jobs = new HashMap<>();
          this.resourceIdJobIdIndex = new HashMap<>();
     }

taskExecutor.start();

public final void start() {
        rpcServer.start();
    }

    protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
        this.rpcService = checkNotNull(rpcService, "rpcService");
        this.endpointId = checkNotNull(endpointId, "endpointId");
        this.rpcServer = rpcService.startServer(this);
        this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
    }

super(rpcService, AkkaRpcServiceUtils.createRandomName(TASK_MANAGER_NAME));

@Override
    public void start() {
        rpcEndpoint.tell(ControlMessages.START, ActorRef.noSender());
    }

@Override
public Receive createReceive() {
    return ReceiveBuilder.create()
        .match(RemoteHandshakeMessage.class, this::handleHandshakeMessage)
        .match(ControlMessages.class, this::handleControlMessage)
        .matchAny(this::handleMessage)
        .build();
}

    @Override
    public void onStart() throws Exception {
        try {
            // 启动相关服务
            startTaskExecutorServices();
        } catch (Exception e) {
            final TaskManagerException exception = new TaskManagerException(String.format("Could not start the TaskExecutor %s", getAddress()), e);
            onFatalError(exception);
            throw exception;
        }
        // 处理注册超时的问题
        startRegistrationTimeout();
    }

    private void startTaskExecutorServices() throws Exception {
        try {
            // start by connecting to the ResourceManager
            resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
            // tell the task slot table who's responsible for the task slot actions
            taskSlotTable.start(new SlotActionsImpl(), getMainThreadExecutor());

            // start the job leader service
            jobLeaderService.start(getAddress(), getRpcService(), haServices, new JobLeaderListenerImpl());

            fileCache = new FileCache(taskManagerConfiguration.getTmpDirectories(), blobCacheService.getPermanentBlobService());
        } catch (Exception e) {
            handleStartTaskExecutorServicesException(e);
        }
    }