前面讲了 SlotManager 的启动流程和 SlotPool 申请资源的源码实现这篇聚焦 SlotManager 分配 Slot 的源码实现。SlotManager 运行在 ResourceManager 内部负责管理所有已注册的 TaskManager 和 Slot接收 SlotPool 的 requestSlot 请求查找空闲 Slot标记分配状态通知 TaskManager 分配确认分配完成。你有没有想过SlotManager 收到请求后是如何查找 FREE Slot 的Slot 状态 FREE → PENDING → ALLOCATED 是如何转换的无空闲 Slot 时是如何处理的Slot 选择策略有哪些这篇从源码级别跟踪 SlotManager 分配 Slot 的完整链路。一、SlotManager 分配 Slot 整体架构下面这张图是 SlotManager 分配 Slot 整体架构包括 ResourceManager、SlotManager 内部组件、TaskManager 和分配流程。SlotManager 运行在 ResourceManager 内部与 TaskManagerManager、ResourceManagerDriver、taskManagerConnections 协作。SlotManager 内部维护三个核心组件taskManagerSlots所有 Slot 映射key 为 SlotIDvalue 为 TaskManagerSlot记录所有已注册的 Slot 及其状态。pendingSlotRequests等待中请求映射key 为 AllocationIDvalue 为 PendingSlotRequest记录正在等待 Slot 可用的请求。slotStatusSyncerSlot 状态同步器Slot 状态变化时同步到 ResourceManager 和 SlotPool。SlotManager 通过 TaskExecutorGateway 与 TaskManager 通信发送 requestSlot RPC 请求。TaskExecutor 收到请求后在本地 SlotTable 中分配 Slot返回 Acknowledge 确认。分配成功后SlotManager 标记 Slot 为 ALLOCATED通知 ResourceManagerRM 向 SlotPool 发送 SlotRequestSuccess 通知。二、SlotManager 分配 Slot 完整流程下面这张图是 SlotManager 分配 Slot 完整流程包括6步分配流程、方法调用链、核心类和常见问题。SlotManager 分配 Slot 分为6步第1步接收 Slot 请求。ResourceManager 收到 SlotPool 的 requestSlot RPC 请求后调用 slotManager.requestSlot(jobId, allocationId, resourceProfile) 方法将请求转发给 SlotManager。SlotManager 验证请求参数检查是否已存在相同 allocationId 的请求避免重复分配。如果已存在相同请求直接返回不重复处理。第2步查找 FREE Slot。SlotManager 调用 findFreeSlot(resourceProfile) 方法遍历所有已注册的 TaskManagerSlot查找状态为 FREE 且满足资源需求CPU、内存、网络的 Slot。查找策略优先选择资源匹配度最高的 Slot其次按 TaskManager 注册顺序轮询支持 Slot 亲和性调度优先选择与输入源同节点的 Slot。第3步标记 PENDING 状态。找到合适的 FREE Slot 后调用 taskManagerSlot.assignPendingAllocation(allocationId, jobId) 方法将 Slot 状态从 FREE 改为 PENDING。该方法有严格的前置条件检查Preconditions.checkState(state FREE)防止非法状态转换。同时创建 PendingSlotRequest记录请求信息和对应的 Slot加入 pendingSlotRequests 映射。第4步通知 TaskManager 分配。SlotManager 通过 taskManagerConnections 获取 TaskExecutorGateway调用 taskExecutorGateway.requestSlot(slotId, jobId, allocationId, targetAddress, timeout) 方法向 TaskManager 发送 Slot 分配请求。TaskExecutor 收到请求后在本地 SlotTable 中分配 Slot返回 Acknowledge 确认。如果 TaskManager 无响应或分配失败SlotManager 会取消 PENDING 状态重新查找其他 Slot。第5步确认 ALLOCATED 状态。TaskExecutor 分配成功后返回 AcknowledgeSlotManager 收到确认后调用 taskManagerSlot.markAllocated() 方法将 Slot 状态从 PENDING 改为 ALLOCATED。该方法同样有前置条件检查state PENDING。状态变化时通知 SlotStatusListener触发 SlotStatusSyncer 同步状态。从 pendingSlotRequests 中移除对应的请求标记分配完成。第6步通知 ResourceManager 成功。SlotManager 分配完成后ResourceManager 向 SlotPool 发送 SlotRequestSuccess(allocationId, slotId) 通知。SlotPool 收到通知后创建 AllocatedSlot加入 allocatedSlots通知等待的 Execution。SchedulerNG 获得 Slot 后通过 TaskExecutorGateway 发送 submitTask 请求TaskExecutor 创建 Task 实例分配到 Slot 上启动执行。三、核心类源码剖析3.1 SlotManagerImpl 核心实现publicclassSlotManagerImplimplementsSlotManager{privatefinalMapSlotID,TaskManagerSlottaskManagerSlots;privatefinalMapAllocationID,PendingSlotRequestpendingSlotRequests;privatefinalSlotStatusSyncerslotStatusSyncer;privatefinalMapResourceID,TaskExecutorGatewaytaskManagerConnections;OverridepublicvoidrequestSlot(JobIDjobId,AllocationIDallocationId,ResourceProfileresourceProfile,StringtargetAddress){// 1. 检查是否已存在相同请求if(pendingSlotRequests.containsKey(allocationId)){return;// 避免重复分配}// 2. 查找 FREE SlotTaskManagerSlotfreeSlotfindFreeSlot(resourceProfile,targetAddress);if(freeSlot!null){// 3. 标记 PENDING 状态freeSlot.assignPendingAllocation(allocationId,jobId);// 4. 创建 PendingSlotRequestPendingSlotRequestpendingRequestnewPendingSlotRequest(allocationId,jobId,resourceProfile,freeSlot.getSlotId());pendingSlotRequests.put(allocationId,pendingRequest);// 5. 通知 TaskManager 分配TaskExecutorGatewaytaskExecutorGatewaytaskManagerConnections.get(freeSlot.getTaskManagerId());taskExecutorGateway.requestSlot(freeSlot.getSlotId(),jobId,allocationId,targetAddress,timeout);}else{// 6. 无 FREE Slot创建 PendingSlotRequest 等待PendingSlotRequestpendingRequestnewPendingSlotRequest(allocationId,jobId,resourceProfile,null);pendingSlotRequests.put(allocationId,pendingRequest);}}privateTaskManagerSlotfindFreeSlot(ResourceProfileresourceProfile,StringtargetAddress){// 1. 优先亲和性查找与 targetAddress 同节点的 FREE Slotif(targetAddress!null){for(TaskManagerSlotslot:taskManagerSlots.values()){if(slot.getState()SlotState.FREEslot.getTaskManagerLocation().getHostname().equals(targetAddress)slot.getResourceProfile().isMatching(resourceProfile)){returnslot;}}}// 2. 资源匹配查找满足资源需求的 FREE Slotfor(TaskManagerSlotslot:taskManagerSlots.values()){if(slot.getState()SlotState.FREEslot.getResourceProfile().isMatching(resourceProfile)){returnslot;}}returnnull;// 无可用 Slot}OverridepublicvoidnotifyAllocationComplete(AllocationIDallocationId,SlotIDslotId){// 1. 查找对应的 PendingSlotRequestPendingSlotRequestpendingRequestpendingSlotRequests.remove(allocationId);if(pendingRequestnull){return;}// 2. 查找对应的 TaskManagerSlotTaskManagerSlotslottaskManagerSlots.get(slotId);if(slotnull){return;}// 3. 标记 ALLOCATED 状态slot.markAllocated();// 4. 通知状态同步器slotStatusSyncer.notifySlotStatusChange(slotId,SlotState.ALLOCATED);}}SlotManagerImpl 是 SlotManager 的核心实现。requestSlot 方法首先检查是否已存在相同请求避免重复分配然后调用 findFreeSlot() 查找 FREE Slot有空闲则标记 PENDING 并通知 TM 分配无空闲则创建 PendingSlotRequest 等待。findFreeSlot 方法优先亲和性调度与 targetAddress 同节点其次资源匹配。notifyAllocationComplete 方法接收 TM 的分配确认标记 ALLOCATED 状态通知状态同步器。3.2 TaskManagerSlot 状态管理publicclassTaskManagerSlot{privatefinalSlotIDslotId;privatefinalResourceIDtaskManagerId;privatefinalTaskManagerLocationtaskManagerLocation;privatefinalResourceProfileresourceProfile;privateSlotStatestate;privateAllocationIDallocationId;privateJobIDjobId;publicTaskManagerSlot(SlotIDslotId,ResourceIDtaskManagerId,TaskManagerLocationtaskManagerLocation,ResourceProfileresourceProfile){this.slotIdslotId;this.taskManagerIdtaskManagerId;this.taskManagerLocationtaskManagerLocation;this.resourceProfileresourceProfile;this.stateSlotState.FREE;}publicvoidassignPendingAllocation(AllocationIDallocationId,JobIDjobId){// 前置条件检查当前状态必须是 FREEPreconditions.checkState(stateSlotState.FREE,Cannot assign pending allocation to slot in state %s,state);this.stateSlotState.PENDING;this.allocationIdallocationId;this.jobIdjobId;}publicvoidmarkAllocated(){// 前置条件检查当前状态必须是 PENDINGPreconditions.checkState(stateSlotState.PENDING,Cannot mark slot as allocated in state %s,state);this.stateSlotState.ALLOCATED;}publicvoidfreeSlot(){// 释放 Slot回到 FREE 状态this.stateSlotState.FREE;this.allocationIdnull;this.jobIdnull;}publicSlotStategetState(){returnstate;}publicSlotIDgetSlotId(){returnslotId;}publicResourceIDgetTaskManagerId(){returntaskManagerId;}publicTaskManagerLocationgetTaskManagerLocation(){returntaskManagerLocation;}publicResourceProfilegetResourceProfile(){returnresourceProfile;}}publicenumSlotState{FREE,// 空闲可分配PENDING,// 分配中等待 TM 确认ALLOCATED// 已分配执行 Task}TaskManagerSlot 表示一个 Slot 实例管理 Slot 的状态转换。Slot 有三种状态FREE空闲可分配、PENDING分配中等待 TM 确认、ALLOCATED已分配执行 Task。assignPendingAllocation 方法将状态从 FREE 改为 PENDING有严格的前置条件检查。markAllocated 方法将状态从 PENDING 改为 ALLOCATED同样有前置条件检查。freeSlot 方法释放 Slot回到 FREE 状态。3.3 PendingSlotRequest 等待请求publicclassPendingSlotRequest{privatefinalAllocationIDallocationId;privatefinalJobIDjobId;privatefinalResourceProfileresourceProfile;privatefinalSlotIDassignedSlot;privatefinallongcreationTime;publicPendingSlotRequest(AllocationIDallocationId,JobIDjobId,ResourceProfileresourceProfile,SlotIDassignedSlot){this.allocationIdallocationId;this.jobIdjobId;this.resourceProfileresourceProfile;this.assignedSlotassignedSlot;this.creationTimeSystem.currentTimeMillis();}publicbooleanisTimeout(longtimeout){returnSystem.currentTimeMillis()-creationTimetimeout;}publicAllocationIDgetAllocationId(){returnallocationId;}publicJobIDgetJobId(){returnjobId;}publicResourceProfilegetResourceProfile(){returnresourceProfile;}publicSlotIDgetAssignedSlot(){returnassignedSlot;}}PendingSlotRequest 封装了等待中的 Slot 请求包含 allocationId、jobId、resourceProfile、assignedSlot已分配的 Slot无空闲时为 null等请求信息。creationTime 记录请求创建时间用于超时检测。isTimeout 方法判断请求是否超时超过 request-timeout默认5分钟未分配则取消请求。3.4 SlotStatusSyncer 状态同步publicclassSlotStatusSyncer{privatefinalResourceManagerGatewayresourceManagerGateway;privatefinalMapSlotID,SlotStateslotStatusCache;publicvoidnotifySlotStatusChange(SlotIDslotId,SlotStatenewState){// 1. 更新状态缓存SlotStateoldStateslotStatusCache.put(slotId,newState);// 2. 状态变化时通知 ResourceManagerif(oldState!newState){resourceManagerGateway.notifySlotStatus(slotId,newState,System.currentTimeMillis());}}publicvoidsyncAllSlotStatus(){// 全量同步所有 Slot 状态for(Map.EntrySlotID,SlotStateentry:slotStatusCache.entrySet()){resourceManagerGateway.notifySlotStatus(entry.getKey(),entry.getValue(),System.currentTimeMillis());}}}SlotStatusSyncer 是 Slot 状态同步器负责将 Slot 状态变化同步到 ResourceManager。notifySlotStatusChange 方法更新状态缓存状态变化时通过 ResourceManagerGateway 通知 RM。syncAllSlotStatus 方法全量同步所有 Slot 状态用于 RM 重新连接后的状态恢复。四、Slot 选择策略与无空闲处理下面这张图是 Slot 选择策略与无空闲处理包括三种选择策略、无空闲处理流程、超时检查、配置和最佳实践。Slot 选择策略SlotManager 分配 Slot 时有三种选择策略资源匹配优先优先选择资源需求匹配度最高的 Slot。比较 Slot 的 ResourceProfile 与请求的 resourceProfile选择 CPU、内存、网络等资源最匹配的 Slot避免资源浪费和资源不足。轮询分配按 TaskManager 注册顺序轮询分配 Slot保证 Slot 在不同 TaskManager 间均匀分布避免单个 TaskManager 负载过高。适用于资源需求相同的 Slot 请求。Slot 亲和性优先选择与输入源同一节点的 Slot提高数据本地性减少网络传输开销。Locality 分为 LOCAL同节点、HOST_SLOT同主机、UNKNOWN无法判断三个级别。无空闲 Slot 处理无 FREE Slot 时SlotManager 创建 PendingSlotRequest 等待同时通知 ResourceManager 通过 ResourceManagerDriver 申请新容器。新 TaskManager 注册后新 Slot 可用SlotManager 分配给等待的 PendingSlotRequest。超过 request-timeout默认5分钟未分配则取消请求通知 SlotPool 分配失败。无空闲处理流程无 FREE SlotfindFreeSlot() 返回 null创建 PendingRequest加入 pendingSlotRequests 等待通知 RM 申请资源ResourceManagerDriver 申请新容器新 TM 注册新 Slot 可用分配等待请求超时取消超过 request-timeout 则取消超时检查机制超时检查通过三个关键参数控制request-timeout默认5分钟Slot 请求超时超过该时间未分配则取消 PendingSlotRequesttaskmanager-timeout默认30秒TaskManager 超时超过该时间无心跳则剔除 TMcheck-interval默认10秒超时检查间隔定期遍历 pendingSlotRequests五、关键配置参数# Slot 请求超时slotmanager.request-timeout:5min# TaskManager 超时slotmanager.taskmanager-timeout:30s# 每个 TaskManager 的 Slot 数taskmanager.numberOfTaskSlots:1# 调度器类型jobmanager.scheduler:adaptive# 最大 Slot 数slotmanager.max-number-of-slots:2147483647# 故障恢复策略jobmanager.execution.failover-strategy:region# 重启策略类型restart-strategy.type:fixed-delay# 固定延迟重启次数restart-strategy.fixed-delay.attempts:3# 固定延迟重启间隔restart-strategy.fixed-delay.delay:10s关键配置说明slotmanager.request-timeoutSlot 请求超时默认5分钟大作业可调大到10-15分钟。slotmanager.taskmanager-timeoutTaskManager 超时默认30秒网络不稳定环境可调大。taskmanager.numberOfTaskSlots每个 TM 的 Slot 数根据 CPU 核数配置通常1-4个。jobmanager.scheduler调度器类型adaptive 支持动态调整并行度。slotmanager.max-number-of-slots最大 Slot 数限制集群可分配的 Slot 总数。jobmanager.execution.failover-strategy故障恢复策略region 只重启受影响的 Task 区域。六、常见问题与最佳实践常见问题无 FREE Slot 可分配原因所有 Slot 已分配或资源不足。解决增加 TM 数量调整并行度等待 Slot 释放检查资源配置。PENDING 状态卡住原因TM 无响应或网络异常分配请求未收到确认。解决检查 TM 心跳确认网络连通调大超时时间查看 TM 日志。状态转换异常原因非法状态转换或并发问题Preconditions.checkState 抛出异常。解决检查日志重启 JM查看 SlotManager 状态确认是否有并发分配。分配超时原因资源不足或 TM 启动慢新容器启动时间长。解决调大 request-timeout增加资源预启动 TM检查 Driver 配置。Slot 亲和性失效原因targetAddress 配置错误或 Slot 不足无法满足本地性要求。解决检查输入源位置配置确保足够 Slot监控 Locality 级别接受 UNKNOWN 降级。最佳实践合理配置 Slot 数根据 CPU 核数配置 numberOfTaskSlots通常1-4个。Slot 数过多会导致每个 Slot 分配的内存和 CPU 减少影响 Task 执行性能过少则资源利用率低。配置合适超时时间大作业可调大 request-timeout 到10-15分钟避免资源申请频繁超时导致作业失败。但不要过大否则资源不足时作业长时间挂起影响集群稳定性。监控 PendingRequest关注等待中的 Slot 请求数量和等待时间及时发现资源不足问题。pendingSlotRequests 持续增长说明资源不足需要增加 TM 或调整并行度。使用亲和性调度配置 targetAddress 提高数据本地性减少网络传输提升作业性能。优先选择 LOCAL 级别的 Slot无法满足时接受 HOST_SLOT 或 UNKNOWN 降级。预启动 TaskManager资源波动大的环境可预启动 TM避免新容器启动慢导致 Slot 分配超时。预启动的 TM 注册后 Slot 立即可用减少等待时间。配置重启策略生产环境配置 fixed-delay 或 failure-rate 重启策略Slot 分配失败时自动重试提高作业稳定性。使用 adaptive 调度器adaptive 调度器支持动态调整并行度根据可用 Slot 数自动调整提高资源利用率和作业稳定性。七、总结Flink SlotManager 分配 Slot 是一个6步流程接收 Slot 请求 → 查找 FREE Slot → 标记 PENDING 状态 → 通知 TaskManager 分配 → 确认 ALLOCATED 状态 → 通知 ResourceManager 成功。核心类包括 SlotManagerImplSlot 管理器接收请求查找和分配 Slot、TaskManagerSlotSlot 实例管理 FREE/PENDING/ALLOCATED 状态转换、PendingSlotRequest待处理请求记录请求信息和超时、SlotStatusSyncerSlot 状态同步器状态变化时同步到 RM。Slot 选择策略有三种资源匹配优先选择 ResourceProfile 匹配度最高的 Slot、轮询分配按 TM 注册顺序轮询保证负载均衡、Slot 亲和性优先选择与输入源同节点的 Slot提高数据本地性。无 FREE Slot 时创建 PendingSlotRequest 等待通知 RM 申请新容器新 TM 注册后分配等待请求超过 request-timeout默认5分钟未分配则取消。理解 SlotManager 分配 Slot 的源码实现有助于排查无 FREE Slot、PENDING 状态卡住、状态转换异常、分配超时、亲和性失效等常见问题为 Flink 作业的稳定运行提供保障。合理配置 Slot 数、超时时间、监控 PendingRequest、使用亲和性调度和预启动 TM可以显著提高 Slot 分配效率和作业稳定性。