Flink的ResourceManager详解（一）
ResourceManager 总结
一、概述
1、ResourceManager 管理 Flink 集群中的计算资源，计算资源主要来自 TaskManager 组件。
2、如果集群采用 Native【本地模式】部署，则 ResourceManager 会动态地向集群资源管理器申请 Container 并启动TaskManager，例如Hadoop Yarn、Kubernetes等。
3、ResourceManager主要接收来自 JobManager 的 SlotRequest 和 TaskManager 的 SlotReport。
二、分类
1、动态资源管理 和 不支持动态资源管理
1）一类支持动态资源管理，例如KubernetesResourceManager、YarnResourceManager及MesosResourceManager
支持动态资源管理的集群类型，可以按需启动TaskManager资源，根据Job所需的资源请求，动态启动TaskManager节点，这种资源管理方式不用担心资源浪费和资源动态伸缩的问题。
实现动态资源管理的ResourceManager需要继承ActiveResourceManager基本实现类。
2）另一类不支持动态资源管理，例如StandaloneResourceManager
2、分类图
三、核心服务
ResourceManagerRuntimeServices 中包含 SlotManager 和 JobLeaderldService 两个主要服务和 HeartbeatService 心跳服务。
1、SlotManager 管理整个集群的 Slot 计算资源，并对 Slot 计算资源进行统一的分配和管理，同时实现了对 TaskManager 信息的注册和管理。
2、JobLeaderldService 通过实现 jobLeaderldListeners 实时监听 JobManager 的运行状态，以获取集群启动的作业对应的 JobLeaderld 信息，防止出现 JobManager 无法连接的情况，用于管理注册的 JobManager 节点，包括对 JobManager 的注册和注销等操作。
3、HeartbeatService 主要通过 TaskManagerHeartbeatListener 和 JobManagerHeartbeatListener 两个监听器收集来自 TaskManager和 JobManager 的心跳信息，以保证整个运行时中各个组件之间能够正常通信。
四、ResourceManager 的初始化和启动
DefaultDispatcherResourceManagerComponentFactory#create 方法
1、初始化 ResourceManager
resourceManager =resourceManagerFactory.createResourceManager(configuration,ResourceID.generate(),rpcService,highAvailabilityServices,heartbeatServices,fatalErrorHandler,new ClusterInformation(hostname, blobServer.getPort()),webMonitorEndpoint.getRestBaseUrl(),metricRegistry,hostname,ioExecutor);
1）创建 ResourceManagerRuntimeServices
1.创建 SlotManager
SlotMatchingStrategy 根据作业中给定的 ResourceProfile 匹配 Slot 计算资源。SlotMatchingStrategy主要分为两种类型：
一种是LeastUtilizationSlotMatchingStrategy，即按照利用率最低原则匹配Slot资源，尽可能保证TaskExecutor上资源的使用率处于比较低的水平，这种策略能够有效降低机器的负载。
另一种是AnyMatchingSlotMatchingStrategy，即直接返回第一个匹配的Slot资源策略。
private static SlotManager createSlotManager(ResourceManagerRuntimeServicesConfiguration configuration,ScheduledExecutor scheduledExecutor,SlotManagerMetricGroup slotManagerMetricGroup) {final SlotManagerConfiguration slotManagerConfiguration =configuration.getSlotManagerConfiguration();if (configuration.isEnableFineGrainedResourceManagement()) {return new FineGrainedSlotManager(scheduledExecutor,slotManagerConfiguration,slotManagerMetricGroup,new DefaultResourceTracker(),new FineGrainedTaskManagerTracker(),new DefaultSlotStatusSyncer(slotManagerConfiguration.getTaskManagerRequestTimeout()),new DefaultResourceAllocationStrategy(SlotManagerUtils.generateTaskManagerTotalResourceProfile(slotManagerConfiguration.getDefaultWorkerResourceSpec()),slotManagerConfiguration.getNumSlotsPerWorker()),Time.milliseconds(REQUIREMENTS_CHECK_DELAY_MS));} else if (configuration.isDeclarativeResourceManagementEnabled()) {return new DeclarativeSlotManager(scheduledExecutor,slotManagerConfiguration,slotManagerMetricGroup,new DefaultResourceTracker(),new DefaultSlotTracker());} else {return new SlotManagerImpl(scheduledExecutor, slotManagerConfiguration, slotManagerMetricGroup);}}
2.创建 JobLeaderIdService
final JobLeaderIdService jobLeaderIdService =new DefaultJobLeaderIdService(highAvailabilityServices, scheduledExecutor, configuration.getJobTimeout());
2）返回创建的 StandaloneResourceManager
return new StandaloneResourceManager(rpcService,resourceId,highAvailabilityServices,heartbeatServices,resourceManagerRuntimeServices.getSlotManager(),ResourceManagerPartitionTrackerImpl::new,resourceManagerRuntimeServices.getJobLeaderIdService(),clusterInformation,fatalErrorHandler,resourceManagerMetricGroup,standaloneClusterStartupPeriodTime,AkkaUtils.getTimeoutAsTime(configuration),ioExecutor);
在 StandaloneResourceManager 构造方法中启动 RpcServer
this.rpcServer = rpcService.startServer(this);
2、启动 ResourceManager
resourceManager.start()->ResourceManager#onStart
ResourceManager#startResourceManagerServices
1）获取 leaderElectionService
leaderElectionService =highAvailabilityServices.getResourceManagerLeaderElectionService();
2）初始化 resourceManagerDriver【ActiveResourceManager需要】
resourceManagerDriver.initialize(this, new GatewayMainThreadExecutor(), ioExecutor);
3）启动 leader 竞选，在 leader 节点启动服务
1.启动心跳服务
在ResourceManager中HeartbeatService的启动方法中，包括了对taskManagerHeartbeatManager和jobManagerHeartbeatManager两个心跳管理服务的启动操作。
而心跳管理服务主要通过TaskManagerHeartbeatListener和JobManagerHeartbeatListener两个监听器收集来自TaskManager和JobManager的心跳信息，以保证整个运行时中各个组件之间能够正常通信。
startHeartbeatServices();
2.启动 slotManager 服务
通过scheduledExecutor线程池启动TaskManager周期性超时检查服务，通过checkTaskManagerTimeouts()方法实现该检查，防止TaskManager长时间掉线等问题。
启动单独的线程对提交的SlotRequest进行周期性超时检查，防止Slot请求超时。
slotManager.start(getFencingToken(), getMainThreadExecutor(), new ResourceActionsImpl());
4）启动 jobLeaderIdService
jobLeaderIdService.start(new JobLeaderIdActionsImpl());