Flink SlotManager 分配 Slot 源码深度剖析：从请求接收到状态转换的完整链路-开发者社区

前面讲了 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 为 SlotID，value 为 TaskManagerSlot，记录所有已注册的 Slot 及其状态。
pendingSlotRequests：等待中请求映射，key 为 AllocationID，value 为 PendingSlotRequest，记录正在等待 Slot 可用的请求。
slotStatusSyncer：Slot 状态同步器，Slot 状态变化时同步到 ResourceManager 和 SlotPool。

SlotManager 通过 TaskExecutorGateway 与 TaskManager 通信，发送 requestSlot RPC 请求。TaskExecutor 收到请求后在本地 SlotTable 中分配 Slot，返回 Acknowledge 确认。分配成功后，SlotManager 标记 Slot 为 ALLOCATED，通知 ResourceManager，RM 向 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 分配成功后返回 Acknowledge，SlotManager 收到确认后调用 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{privatefinalMap<SlotID,TaskManagerSlot>taskManagerSlots;privatefinalMap<AllocationID,PendingSlotRequest>pendingSlotRequests;privatefinalSlotStatusSyncerslotStatusSyncer;privatefinalMap<ResourceID,TaskExecutorGateway>taskManagerConnections;@OverridepublicvoidrequestSlot(JobIDjobId,AllocationIDallocationId,ResourceProfileresourceProfile,StringtargetAddress){// 1. 检查是否已存在相同请求if(pendingSlotRequests.containsKey(allocationId)){return;// 避免重复分配}// 2. 查找 FREE SlotTaskManagerSlotfreeSlot=findFreeSlot(resourceProfile,targetAddress);if(freeSlot!=null){// 3. 标记 PENDING 状态freeSlot.assignPendingAllocation(allocationId,jobId);// 4. 创建 PendingSlotRequestPendingSlotRequestpendingRequest=newPendingSlotRequest(allocationId,jobId,resourceProfile,freeSlot.getSlotId());pendingSlotRequests.put(allocationId,pendingRequest);// 5. 通知 TaskManager 分配TaskExecutorGatewaytaskExecutorGateway=taskManagerConnections.get(freeSlot.getTaskManagerId());taskExecutorGateway.requestSlot(freeSlot.getSlotId(),jobId,allocationId,targetAddress,timeout);}else{// 6. 无 FREE Slot，创建 PendingSlotRequest 等待PendingSlotRequestpendingRequest=newPendingSlotRequest(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.FREE&&slot.getTaskManagerLocation().getHostname().equals(targetAddress)&&slot.getResourceProfile().isMatching(resourceProfile)){returnslot;}}}// 2. 资源匹配：查找满足资源需求的 FREE Slotfor(TaskManagerSlotslot:taskManagerSlots.values()){if(slot.getState()==SlotState.FREE&&slot.getResourceProfile().isMatching(resourceProfile)){returnslot;}}returnnull;// 无可用 Slot}@OverridepublicvoidnotifyAllocationComplete(AllocationIDallocationId,SlotIDslotId){// 1. 查找对应的 PendingSlotRequestPendingSlotRequestpendingRequest=pendingSlotRequests.remove(allocationId);if(pendingRequest==null){return;}// 2. 查找对应的 TaskManagerSlotTaskManagerSlotslot=taskManagerSlots.get(slotId);if(slot==null){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.slotId=slotId;this.taskManagerId=taskManagerId;this.taskManagerLocation=taskManagerLocation;this.resourceProfile=resourceProfile;this.state=SlotState.FREE;}publicvoidassignPendingAllocation(AllocationIDallocationId,JobIDjobId){// 前置条件检查：当前状态必须是 FREEPreconditions.checkState(state==SlotState.FREE,"Cannot assign pending allocation to slot in state %s",state);this.state=SlotState.PENDING;this.allocationId=allocationId;this.jobId=jobId;}publicvoidmarkAllocated(){// 前置条件检查：当前状态必须是 PENDINGPreconditions.checkState(state==SlotState.PENDING,"Cannot mark slot as allocated in state %s",state);this.state=SlotState.ALLOCATED;}publicvoidfreeSlot(){// 释放 Slot，回到 FREE 状态this.state=SlotState.FREE;this.allocationId=null;this.jobId=null;}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.allocationId=allocationId;this.jobId=jobId;this.resourceProfile=resourceProfile;this.assignedSlot=assignedSlot;this.creationTime=System.currentTimeMillis();}publicbooleanisTimeout(longtimeout){returnSystem.currentTimeMillis()-creationTime>timeout;}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;privatefinalMap<SlotID,SlotState>slotStatusCache;publicvoidnotifySlotStatusChange(SlotIDslotId,SlotStatenewState){// 1. 更新状态缓存SlotStateoldState=slotStatusCache.put(slotId,newState);// 2. 状态变化时通知 ResourceManagerif(oldState!=newState){resourceManagerGateway.notifySlotStatus(slotId,newState,System.currentTimeMillis());}}publicvoidsyncAllSlotStatus(){// 全量同步所有 Slot 状态for(Map.Entry<SlotID,SlotState>entry: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 Slot：findFreeSlot() 返回 null
创建 PendingRequest：加入 pendingSlotRequests 等待
通知 RM 申请资源：ResourceManagerDriver 申请新容器
新 TM 注册：新 Slot 可用，分配等待请求
超时取消：超过 request-timeout 则取消

超时检查机制

超时检查通过三个关键参数控制：

request-timeout（默认5分钟）：Slot 请求超时，超过该时间未分配则取消 PendingSlotRequest
taskmanager-timeout（默认30秒）：TaskManager 超时，超过该时间无心跳则剔除 TM
check-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-timeout：Slot 请求超时，默认5分钟，大作业可调大到10-15分钟。
slotmanager.taskmanager-timeout：TaskManager 超时，默认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 成功。核心类包括 SlotManagerImpl（Slot 管理器，接收请求，查找和分配 Slot）、TaskManagerSlot（Slot 实例，管理 FREE/PENDING/ALLOCATED 状态转换）、PendingSlotRequest（待处理请求，记录请求信息和超时）、SlotStatusSyncer（Slot 状态同步器，状态变化时同步到 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 分配效率和作业稳定性。