启动脚本分析

TaskManager的启动方式为taskmanager.sh start。因此我们如果想要知道程序入口类，必须首先分析这个脚本。
这个脚本比较长，我们重点关注如下片段：

  1. # ...
  2. # 设置ENTRYPOINT变量值为taskexecutor
  3. ENTRYPOINT=taskexecutor
  4. # ...
  5. if [[ $STARTSTOP == "start-foreground" ]]; then
  6. exec "${FLINK_BIN_DIR}"/flink-console.sh $ENTRYPOINT "${ARGS[@]}"
  7. else
  8. if [[ $FLINK_TM_COMPUTE_NUMA == "false" ]]; then
  9. # Start a single TaskManager
  10. # 我们关注这里，不使用NUMA方式后台启动
  11. "${FLINK_BIN_DIR}"/flink-daemon.sh $STARTSTOP $ENTRYPOINT "${ARGS[@]}"
  12. else
  13. # Example output from `numactl --show` on an AWS c4.8xlarge:
  14. # policy: default
  15. # preferred node: current
  16. # physcpubind: 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35
  17. # cpubind: 0 1
  18. # nodebind: 0 1
  19. # membind: 0 1
  20. read -ra NODE_LIST <<< $(numactl --show | grep "^nodebind: ")
  21. for NODE_ID in "${NODE_LIST[@]:1}"; do
  22. # Start a TaskManager for each NUMA node
  23. numactl --membind=$NODE_ID --cpunodebind=$NODE_ID -- "${FLINK_BIN_DIR}"/flink-daemon.sh $STARTSTOP $ENTRYPOINT "${ARGS[@]}"
  24. done
  25. fi
  26. fi

通过分析上面的脚本，我们看到如果没有使用start-foreground(前台模式启动)，实际上调用的是flink-daemon.sh脚本。此脚本的使用方式和参数示例为：

  1. flink-daemon.sh (start|stop|stop-all) (taskexecutor|zookeeper|historyserver|standalonesession|standalonejob) [args]

为了弄清楚Java的入口类，我们接着分析flink-daemon.sh，发现如下片段：

  1. # ...
  2. STARTSTOP=$1
  3. # 经上面分析可知，通过taskmanager.sh执行，DAEMON的值为taskexecutor
  4. DAEMON=$2
  5. # ...
  6. case $DAEMON in
  7.     (taskexecutor)
  8.         CLASS_TO_RUN=org.apache.flink.runtime.taskexecutor.TaskManagerRunner
  9.     ;;
  11.     (zookeeper)
  12.         CLASS_TO_RUN=org.apache.flink.runtime.zookeeper.FlinkZooKeeperQuorumPeer
  13.     ;;
  15.     (historyserver)
  16.         CLASS_TO_RUN=org.apache.flink.runtime.webmonitor.history.HistoryServer
  17.     ;;
  19.     (standalonesession)
  20.         CLASS_TO_RUN=org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint
  21.     ;;
  23.     (standalonejob)
  24.         CLASS_TO_RUN=org.apache.flink.container.entrypoint.StandaloneApplicationClusterEntryPoint
  25.     ;;
  27.     (*)
  28.         echo "Unknown daemon '${DAEMON}'. $USAGE."
  29.         exit 1
  30.     ;;
  31. esac
  32. # ...

不难发现，DAEMON变量的值为taskexecutor，实际的入口类为org.apache.flink.runtime.taskexecutor.TaskManagerRunner。TaskManagerRunner就是TaskManager启动的入口类。

TaskManagerRunner

TaskManagerRunner是TaskManager在yarn模式和standalone模式下的启动类。
我们查看下main方法：

  1. public static void main(String[] args) throws Exception {
  2.     // startup checks and logging
  3.     // 日志打印环境信息
  4.     EnvironmentInformation.logEnvironmentInfo(LOG, "TaskManager", args);
  6.     // 注册信号处理句柄
  7.     // 在Linux下响应TERM，HUP和INT
  8.     SignalHandler.register(LOG);
  10.     // 增加shutdownlook，在JVM关闭之前回调
  11.     // 让JVM关闭延迟5秒钟
  12.     JvmShutdownSafeguard.installAsShutdownHook(LOG);
  14.     // 获取并打印最大打开文件句柄数限制
  15.     long maxOpenFileHandles = EnvironmentInformation.getOpenFileHandlesLimit();
  17.     if (maxOpenFileHandles != -1L) {
  18.         LOG.info("Maximum number of open file descriptors is {}.", maxOpenFileHandles);
  19.     } else {
  20.         LOG.info("Cannot determine the maximum number of open file descriptors");
  21.     }
  22.     // 启动TaskManager
  23.     runTaskManagerSecurely(args);
  24.     }

继续跟踪runTaskManagerSecurely方法，内容如下：

  1. public static void runTaskManagerSecurely(String[] args) {
  2.     try {
  3.         // // 读取flink-conf.yaml和命令行传入的动态参数，作为配置信息
  4.         Configuration configuration = loadConfiguration(args);
  5.         // 继续调用runTaskManagerSecurely
  6.         runTaskManagerSecurely(configuration);
  7.     } catch (Throwable t) {
  8.         final Throwable strippedThrowable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
  9.         LOG.error("TaskManager initialization failed.", strippedThrowable);
  10.         System.exit(STARTUP_FAILURE_RETURN_CODE);
  11.     }
  12. }

继续跟踪runTaskManagerSecurely方法，内容如下：

  1. public static void runTaskManagerSecurely(Configuration configuration) throws Exception {
  3.     // 替换掉配置文件中包含优雅退出的指令
  4.     replaceGracefulExitWithHaltIfConfigured(configuration);
  6.     // 从配置文件中创建插件管理器
  7.     final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(configuration);
  9.     // 初始化共享文件系统配置(FileSystemFactory)，比如HDFS
  10.     FileSystem.initialize(configuration, pluginManager);
  12.     // 读取安全相关配置，包含flink，JAAS，Hadoop和Zookeeper的安全配置
  13.     SecurityUtils.install(new SecurityConfiguration(configuration));
  15.     // 以安全认证环境下调用runTaskManager
  16.     SecurityUtils.getInstalledContext().runSecured(() -> {
  17.         runTaskManager(configuration, pluginManager);
  18.         return null;
  19.     });
  20. }

该方法载入了Flink的主配置文件，初始化了文件系统和服务安全配置。启动TaskManager的方法在runTaskManager。如下所示：

  1. public static void runTaskManager(Configuration configuration, PluginManager pluginManager) throws Exception {
  2.     // 创建一个TaskManagerRunner
  3.     final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, 
  4.                                                                       TaskManagerRunner::createTaskExecutorService);
  5.     // 调用start
  6.     taskManagerRunner.start();
  7. }

首先我们分析下构造函数：

  1. public TaskManagerRunner(
  2.         Configuration configuration,
  3.         PluginManager pluginManager,
  4.         TaskExecutorServiceFactory taskExecutorServiceFactory) throws Exception {
  5.     this.configuration = checkNotNull(configuration);
  7.     // 获取Akka超时时间
  8.     timeout = AkkaUtils.getTimeoutAsTime(configuration);
  10.     // 创建一个task manager的线程池
  11.     // corePoolSize和机器CPU数量一致
  12.     this.executor = java.util.concurrent.Executors.newScheduledThreadPool(
  13.         Hardware.getNumberCPUCores(),
  14.         new ExecutorThreadFactory("taskmanager-future"));
  16.     // 创建高可用服务
  17.     // 负责选举JobManager和ResourceManager和获取leader信息
  18.     highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
  19.         configuration,
  20.         executor,
  21.         HighAvailabilityServicesUtils.AddressResolution.NO_ADDRESS_RESOLUTION);
  23.     JMXService.startInstance(configuration.getString(JMXServerOptions.JMX_SERVER_PORT));
  25.     // 创建RPC服务，和其他Flink进程互相通信的时候使用
  26.     // 用于连接一个RpcEndpoint，连接成功之后返回一个RpcGateway
  27.     rpcService = createRpcService(configuration, highAvailabilityServices);
  29.     this.resourceId = getTaskManagerResourceID(configuration, rpcService.getAddress(), rpcService.getPort());
  31.     // 创建心跳服务
  32.     HeartbeatServices heartbeatServices = HeartbeatServices.fromConfiguration(configuration);
  34.     // 创建监控指标注册
  35.     // 用于记录所有的metrics，连接MetricGroup和MetricReporter
  36.     metricRegistry = new MetricRegistryImpl(
  37.         MetricRegistryConfiguration.fromConfiguration(configuration),
  38.         ReporterSetup.fromConfiguration(configuration, pluginManager));
  40.     // 开启metrics查询服务
  41.     // 以key-value方式返回Flink中已注册的metrics
  42.     final RpcService metricQueryServiceRpcService = MetricUtils.startRemoteMetricsRpcService(configuration, rpcService.getAddress());
  43.     metricRegistry.startQueryService(metricQueryServiceRpcService, resourceId);
  45.     // 创建BlobCache服务
  46.     blobCacheService = new BlobCacheService(
  47.         configuration, highAvailabilityServices.createBlobStore(), null
  48.     );
  50.     // 创建外部资源信息Provider
  51.     final ExternalResourceInfoProvider externalResourceInfoProvider =
  52.         ExternalResourceUtils.createStaticExternalResourceInfoProvider(
  53.             ExternalResourceUtils.getExternalResourceAmountMap(configuration),
  54.             ExternalResourceUtils.externalResourceDriversFromConfig(configuration, pluginManager));
  55.     // 启动task manager
  56.     // 稍后分析
  57.     taskExecutorService = taskExecutorServiceFactory.createTaskExecutor(
  58.         this.configuration,
  59.         this.resourceId,
  60.         rpcService,
  61.         highAvailabilityServices,
  62.         heartbeatServices,
  63.         metricRegistry,
  64.         blobCacheService,
  65.         false,
  66.         externalResourceInfoProvider,
  67.         this);
  69.     this.terminationFuture = new CompletableFuture<>();
  70.     this.shutdown = false;
  71.     handleUnexpectedTaskExecutorServiceTermination();
  73.     MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
  74. }

在分析了层层封装之后，createTaskExecutor方法后终于找到startTaskManager方法。该方法创建出一个TaskExecutor对象，如下所示：

  1. public static TaskExecutor startTaskManager(
  2.         Configuration configuration,
  3.         ResourceID resourceID,
  4.         RpcService rpcService,
  5.         HighAvailabilityServices highAvailabilityServices,
  6.         HeartbeatServices heartbeatServices,
  7.         MetricRegistry metricRegistry,
  8.         BlobCacheService blobCacheService,
  9.         boolean localCommunicationOnly,
  10.         ExternalResourceInfoProvider externalResourceInfoProvider,
  11.         FatalErrorHandler fatalErrorHandler) throws Exception {
  13.     checkNotNull(configuration);
  14.     checkNotNull(resourceID);
  15.     checkNotNull(rpcService);
  16.     checkNotNull(highAvailabilityServices);
  18.     LOG.info("Starting TaskManager with ResourceID: {}", resourceID.getStringWithMetadata());
  20.     // 获取外部访问地址
  21.     String externalAddress = rpcService.getAddress();
  23.     // 获取task executor资源详情
  24.     // 包含CPU核数，task堆内存，tsk堆外内存和managed内存
  25.     final TaskExecutorResourceSpec taskExecutorResourceSpec = TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);
  27.     // 获取Task Manager服务配置
  28.     TaskManagerServicesConfiguration taskManagerServicesConfiguration =
  29.         TaskManagerServicesConfiguration.fromConfiguration(
  30.             configuration,
  31.             resourceID,
  32.             externalAddress,
  33.             localCommunicationOnly,
  34.             taskExecutorResourceSpec);
  36.     // 创建task manager的MetricGroup
  37.     Tuple2<TaskManagerMetricGroup, MetricGroup> taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
  38.         metricRegistry,
  39.         externalAddress,
  40.         resourceID,
  41.         taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
  43.     // 创建用于IO任务的线程池
  44.     final ExecutorService ioExecutor = Executors.newFixedThreadPool(
  45.         taskManagerServicesConfiguration.getNumIoThreads(),
  46.         new ExecutorThreadFactory("flink-taskexecutor-io"));
  48.     // 创建Task Manager服务，是其他多种资源或服务的容器
  49.     // 它涉及的服务也非常多，稍后我们单独分析
  50.     TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
  51.         taskManagerServicesConfiguration,
  52.         blobCacheService.getPermanentBlobService(),
  53.         taskManagerMetricGroup.f1,
  54.         ioExecutor,
  55.         fatalErrorHandler);
  57.     MetricUtils.instantiateFlinkMemoryMetricGroup(
  58.         taskManagerMetricGroup.f1,
  59.         taskManagerServices.getTaskSlotTable(),
  60.         taskManagerServices::getManagedMemorySize);
  62.     // 创建task manager的配置
  63.     TaskManagerConfiguration taskManagerConfiguration =
  64.         TaskManagerConfiguration.fromConfiguration(configuration, taskExecutorResourceSpec, externalAddress);
  66.     // 获取metrics查询服务的地址
  67.     String metricQueryServiceAddress = metricRegistry.getMetricQueryServiceGatewayRpcAddress();
  69.     // 创建task executor
  70.     return new TaskExecutor(
  71.         rpcService,
  72.         taskManagerConfiguration,
  73.         highAvailabilityServices,
  74.         taskManagerServices,
  75.         externalResourceInfoProvider,
  76.         heartbeatServices,
  77.         taskManagerMetricGroup.f0,
  78.         metricQueryServiceAddress,
  79.         blobCacheService,
  80.         fatalErrorHandler,
  81.         new TaskExecutorPartitionTrackerImpl(taskManagerServices.getShuffleEnvironment()),
  82.         createBackPressureSampleService(configuration, rpcService.getScheduledExecutor()));
  83. }

到这里为止，一个完整的TaskExecutor创建完毕。创建的过程涉及到了很多相关服务器的初始化，稍后在文末以脑图形式为大家总结。

TaskManagerServices的fromConfiguration方法

TaskManagerServices是一系列TaskManager服务的容器，包含内存控制器，IO控制器，Shuffle环境等。各个服务的用途计划在后续博客中介绍。
这里我们重点关注它的fromConfiguration方法。如下所示：

  1. /**
  2.  * Creates and returns the task manager services.
  3.  *
  4.  * @param taskManagerServicesConfiguration task manager configuration
  5.  * @param permanentBlobService permanentBlobService used by the services
  6.  * @param taskManagerMetricGroup metric group of the task manager
  7.  * @param ioExecutor executor for async IO operations
  8.  * @param fatalErrorHandler to handle class loading OOMs
  9.  * @return task manager components
  10.  * @throws Exception
  11.  */
  12. public static TaskManagerServices fromConfiguration(
  13.         TaskManagerServicesConfiguration taskManagerServicesConfiguration,
  14.         PermanentBlobService permanentBlobService,
  15.         MetricGroup taskManagerMetricGroup,
  16.         ExecutorService ioExecutor,
  17.         FatalErrorHandler fatalErrorHandler) throws Exception {
  19.     // pre-start checks
  20.     // 检查temp dir（yarn模式为local_dirs，standalone模式为java.io.tmpdir）目录是否存在，如果不存在会创建文件夹
  21.     // 是否temp dir是否是目录，是否可写入
  22.     checkTempDirs(taskManagerServicesConfiguration.getTmpDirPaths());
  24.     // 创建TaskEventDispatcher
  25.     // 任务事件派发器，用于消费任务向生产任务发送TaskEvent
  26.     final TaskEventDispatcher taskEventDispatcher = new TaskEventDispatcher();
  28.     // 创建异步IO管理器
  29.     // 根据配置的tmp dir的个数，创建对应数量的读写线程，负责异步读写数据
  30.     // start the I/O manager, it will create some temp directories.
  31.     final IOManager ioManager = new IOManagerAsync(taskManagerServicesConfiguration.getTmpDirPaths());
  33.     // 创建ShuffleEnvironment
  34.     // 负责在本地提供一个shuffle环境，使用memory segment作为数据存储
  35.     // 可以创建数据写入端ResultPartitionWriter和数据消费端InputGate
  36.     final ShuffleEnvironment<?, ?> shuffleEnvironment = createShuffleEnvironment(
  37.         taskManagerServicesConfiguration,
  38.         taskEventDispatcher,
  39.         taskManagerMetricGroup,
  40.         ioExecutor);
  42.     // 启动ShuffleManager
  43.     // 返回TaskManager的数据端口（taskmanager.data.port）
  44.     final int listeningDataPort = shuffleEnvironment.start();
  46.     // 创建key value状态存储服务
  47.     final KvStateService kvStateService = KvStateService.fromConfiguration(taskManagerServicesConfiguration);
  48.     kvStateService.start();
  50.     final UnresolvedTaskManagerLocation unresolvedTaskManagerLocation = new UnresolvedTaskManagerLocation(
  51.         taskManagerServicesConfiguration.getResourceID(),
  52.         taskManagerServicesConfiguration.getExternalAddress(),
  53.         // we expose the task manager location with the listening port
  54.         // iff the external data port is not explicitly defined
  55.         taskManagerServicesConfiguration.getExternalDataPort() > 0 ?
  56.             taskManagerServicesConfiguration.getExternalDataPort() :
  57.             listeningDataPort);
  59.     // 创建广播变量管理器
  60.     final BroadcastVariableManager broadcastVariableManager = new BroadcastVariableManager();
  62.     // 创建TaskSlotTable，维护task和slot的分配关系
  63.     final TaskSlotTable<Task> taskSlotTable = createTaskSlotTable(
  64.         taskManagerServicesConfiguration.getNumberOfSlots(),
  65.         taskManagerServicesConfiguration.getTaskExecutorResourceSpec(),
  66.         taskManagerServicesConfiguration.getTimerServiceShutdownTimeout(),
  67.         taskManagerServicesConfiguration.getPageSize(),
  68.         ioExecutor);
  70.     final JobTable jobTable = DefaultJobTable.create();
  72.     // 创建Job leader服务。Job leader是领导一个job的job manager。
  73.     // 一旦某个job manager获得leader角色，或者失去leader状态，会通知JobLeaderListener，位于TaskExecutor.java中
  74.     final JobLeaderService jobLeaderService = new DefaultJobLeaderService(unresolvedTaskManagerLocation, taskManagerServicesConfiguration.getRetryingRegistrationConfiguration());
  76.     // 读取本地状态保存根路径
  77.     // taskmanager.state.local.root-dirs
  78.     final String[] stateRootDirectoryStrings = taskManagerServicesConfiguration.getLocalRecoveryStateRootDirectories();
  80.     final File[] stateRootDirectoryFiles = new File[stateRootDirectoryStrings.length];
  82.     // 创建目录
  83.     for (int i = 0; i < stateRootDirectoryStrings.length; ++i) {
  84.         stateRootDirectoryFiles[i] = new File(stateRootDirectoryStrings[i], LOCAL_STATE_SUB_DIRECTORY_ROOT);
  85.     }
  87.     // 创建任务状态管理器
  88.     final TaskExecutorLocalStateStoresManager taskStateManager = new TaskExecutorLocalStateStoresManager(
  89.         taskManagerServicesConfiguration.isLocalRecoveryEnabled(),
  90.         stateRootDirectoryFiles,
  91.         ioExecutor);
  93.     final boolean failOnJvmMetaspaceOomError =
  94.         taskManagerServicesConfiguration.getConfiguration().getBoolean(CoreOptions.FAIL_ON_USER_CLASS_LOADING_METASPACE_OOM);
  95.     final boolean checkClassLoaderLeak =
  96.         taskManagerServicesConfiguration.getConfiguration().getBoolean(CoreOptions.CHECK_LEAKED_CLASSLOADER);
  97.     final LibraryCacheManager libraryCacheManager = new BlobLibraryCacheManager(
  98.         permanentBlobService,
  99.         BlobLibraryCacheManager.defaultClassLoaderFactory(
  100.             taskManagerServicesConfiguration.getClassLoaderResolveOrder(),
  101.             taskManagerServicesConfiguration.getAlwaysParentFirstLoaderPatterns(),
  102.             failOnJvmMetaspaceOomError ? fatalErrorHandler : null,
  103.             checkClassLoaderLeak));
  105.     // 构建TaskManagerServices
  106.     return new TaskManagerServices(
  107.         unresolvedTaskManagerLocation,
  108.         taskManagerServicesConfiguration.getManagedMemorySize().getBytes(),
  109.         ioManager,
  110.         shuffleEnvironment,
  111.         kvStateService,
  112.         broadcastVariableManager,
  113.         taskSlotTable,
  114.         jobTable,
  115.         jobLeaderService,
  116.         taskStateManager,
  117.         taskEventDispatcher,
  118.         ioExecutor,
  119.         libraryCacheManager);
  120. }

TaskManager的启动

以上只是相关服务的创建逻辑，服务启动的逻辑位于start方法中。
TaskManagerRunner的start方法如下所示：

附录

TaskExecutor中的重要服务脑图

这些服务的作用在本人后续博客中计划陆续分析。
image.png