标签：slot 启动 Flink TaskManager 源码 TaskExecutor 节点

taskManager是flink的worker节点，负责slot的资源管理和task执行 一个taskManager就是一台服务器的抽象 TaskManager基本资源单位是slot，一个作业的task会部署在一个TM的slot上运行，TM会负责维护本地的slot资源列表，并与Master和JobManager进行通信 启动主类：TaskManagerRunner TaskManagerRunner.main() 启动函数：runTaskManagerProcessSecurely(args); Flink集群主节点和从节点，每个节点都有一个全局的唯一ID，叫ResourceID

1. load解析main方法参数和flink-conf.yaml配置信息
2. runTaskManagerProcessSecurely
   1. 在主节点启动的时候启动了PluginManager
   2. 然后再runTaskManager里面通过线程来启动TaskManager

      SecurityUtils.getInstalledContext()
              .runSecured(() -> runTaskManager(configuration, pluginManager));

       

   3. 构建tm实例，taskManagerRunner是standalone下TaskManager的可执行入口点。构造相关组件（network，IO，memoryManager，RPCService等）并启动
      1. New TaskManagerRunner初始化各种基础服务组件
         1. 初始化各种组件
         2. WebMonitorEndpoint
         3. Dispatcher
         4. resourceManager
      2. start方法是发送一个start消息确认启动

     TaskManager启动分三件事

   1.   启动基础服务

      1. 启动7大基础服务

   2.   初始化taskManagerService

      1. startTaskManager
          启动中启动了很多服务，同时也启动了TaskExecutor
      1. 线程池，处理回调
      1. Flink中大量的异步编程
      2. CompletableFeature大量提交请求的执行和回调的执行都是由线程池执行
      3. feature.xxx(()->xxx(),executor)
      4. 定时的线程池(NewScheduledThreadPool):创建一个线程池，它可安排在延迟时间后运行命令或者定期地执行
      2. HA高可用服务，用zookeeper来做，在flink-conf.yaml中配置high-availability = zookeeper模式

         highAvailabilityServices =
                 HighAvailabilityServicesUtils.createHighAvailabilityServices(
                         configuration,
                         executor,
                         AddressResolution.NO_ADDRESS_RESOLUTION,
                         rpcSystem,
                         this);

          

         
      3. createRpcService
         1. createAndStart初始化actor System
         2. 使用了代理模式
            1. 代理模式是一种结构型设计模式，它允许一个对象代表另一个对象，从而控制对这个对象的访问。在Flink的RPC服务中，代理模式被用于创建远程调用的代理对象，以便在客户端和服务器之间进行通信。
            2. 具体来说，在Flink的RPC框架中，RpcService负责创建和启动RpcEndpoint组件的RpcServer，并且提供了与远程RpcServer建立连接的能力。当调用RpcService的connect方法时，它会创建一个代理对象，这个代理对象实现了RpcGateway接口，用于与远程的RpcEndpoint进行通信。
            3. 当一个TaskManager需要与ResourceManager通信时，它会通过RpcService创建一个连接。RpcService会使用Akka框架创建一个ActorRef，然后通过这个ActorRef创建一个AkkaInvocationHandler或者FencedAkkaInvocationHandler。这个InvocationHandler会被用来创建一个RpcGateway的代理对象，该代理对象可以在本地代表远程的RpcEndpoint
      4. heartbeat服务
         1. taskExecutor心跳组件启动，跟resourceManager维持心跳
         2. jobMaster心跳组件启动，跟resourceManager维持心跳
         3. interval=10s并且timeout=50s，也就是连续5次心跳没有就挂了，在flink-conf.yaml设置
      5. metrics监控服务
      6. Blob资源删除服务：两个定时任务，定时检查，删除过期的job资源文件，通过引用计数法来判断文件是否过期
         1. 主节点启动BlobServer
         2. 从节点启动BlobCacheService
   1. 在startTaskManager的最后，初始化TaskExecutor
      1. Task资源

         final TaskExecutorResourceSpec taskExecutorResourceSpec =
                 TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);

      2. 1台物理节点的资源：cpu，memory。network
   2. taskManagerService
      1. 初始化了很多的taskManager要运行过程中需要的服务，真正需要的用来对外提供服务的组件

      TaskManagerServices taskManagerServices =
              TaskManagerServices.fromConfiguration(
                      taskManagerServicesConfiguration,
                      taskExecutorBlobService.getPermanentBlobService(),
                      taskManagerMetricGroup.f1,
                      ioExecutor,
                      fatalErrorHandler,
                      workingDirectory);

       

      
      1. 里面有个shuffleEnvironment，上下游StreamTask有shuffle动作，在过程中，许哟啊很组件工作，这个组件为shuffle提供创建组件的环境支持
      2. TaskSlotTable<Task> taskSlotTable
         1. 哪些task在哪些slot里执行，使用table关联
         2. 当一个slot被调度执行了一个task，就会生成taskslot对象
         3. 一个task是由一个线程执行
      3. Flink：resourceManager+taskManager（slot管理+task执行）
   3. 方法返回TaskExecutor，初始化taskManager最重要的目标就是启动taskExecutor

      return new TaskExecutor(
              rpcService,
              taskManagerConfiguration,
              highAvailabilityServices,
              taskManagerServices,
              externalResourceInfoProvider,
              heartbeatServices,
              taskManagerMetricGroup.f0,
              metricQueryServiceAddress,
              taskExecutorBlobService,
              fatalErrorHandler,
              new TaskExecutorPartitionTrackerImpl(taskManagerServices.getShuffleEnvironment()),
              delegationTokenReceiverRepository);

       

   4. 额外
   Flink集群主从架构：JobManager，TaskManager ResourceManager（心跳，线程池）+taskExecutor（slot管理，task执行） 不管是主节点JobManager还是从节点TaskManager除了关于资源管理和调度意外，还需要其他的服务 JobManager是逻辑概念上的主节点，实际上类叫clusterEntryPoint，然后具体叫JobMaster

Standalone

1个Yarn Container JobManager 启动ResourceManager 启动Dispatcher 其他的 Yarn Container TaskExecutor 应用程序提交的时候，启动一个JobMaster，再去调度StreamTask执行，向JobMaster汇报心跳 Dispatcher调度一个JobMaster，

 

3. TaskManager/TaskExecutor注册和心跳

总结：taskManager是一个逻辑抽象，代表一台服务器，启动必然会包含一些服务，另外包含一个TaskExecutor存在于内部，真实ide帮助Task Manager完成各种核心操作：

1. 提交task
2. 申请和释放slot

创建TaskManager实际上返回的是TaskExecutor TaskExecutor本身是RpcEndpoint的子类

1. TaskExecutor的初始化：
   1. 初始化2个心跳管理器，jobManagerHeartbeat resourceManagerHeartbeat
   2. hardware Description把硬件资源抽象成一个对象

      this.hardwareDescription =
              HardwareDescription.extractFromSystem(taskExecutorServices.getManagedMemorySize());

       

2. onStart方法
   1. 继承的RpcEndpoint，所以需要执行
   2. 启动taskExecutorService
      1. 监控ResourceManager
         1. 连接ResourceManager
         2. 注册
         3. 维持心跳
         4. 当前TaskExecutor会监控RM的变更
      2. 启动TaskSlotTable服务
      3. 监控JobMaster
      4. 启动FileCache

标签：slot,启动,Flink,TaskManager,源码,TaskExecutor,节点
From： https://www.cnblogs.com/ak918xp/p/18130269