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

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

前面讲了 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 核心实现

public class SlotManagerImpl implements SlotManager {
private final Map<SlotID, TaskManagerSlot> taskManagerSlots;
private final Map<AllocationID, PendingSlotRequest> pendingSlotRequests;
private final SlotStatusSyncer slotStatusSyncer;
private final Map<ResourceID, TaskExecutorGateway> taskManagerConnections;

@Override
public void requestSlot(
JobID jobId,
AllocationID allocationId,
ResourceProfile resourceProfile,
String targetAddress) {

// 1. 检查是否已存在相同请求
if (pendingSlotRequests.containsKey(allocationId)) {
return; // 避免重复分配
}

// 2. 查找 FREE Slot
TaskManagerSlot freeSlot = findFreeSlot(resourceProfile, targetAddress);
if (freeSlot != null) {
// 3. 标记 PENDING 状态
freeSlot.assignPendingAllocation(allocationId, jobId);

// 4. 创建 PendingSlotRequest
PendingSlotRequest pendingRequest = new PendingSlotRequest(
allocationId, jobId, resourceProfile, freeSlot.getSlotId());
pendingSlotRequests.put(allocationId, pendingRequest);

// 5. 通知 TaskManager 分配
TaskExecutorGateway taskExecutorGateway =
taskManagerConnections.get(freeSlot.getTaskManagerId());
taskExecutorGateway.requestSlot(
freeSlot.getSlotId(),
jobId,
allocationId,
targetAddress,
timeout);
} else {
// 6. 无 FREE Slot，创建 PendingSlotRequest 等待
PendingSlotRequest pendingRequest = new PendingSlotRequest(
allocationId, jobId, resourceProfile, null);
pendingSlotRequests.put(allocationId, pendingRequest);
}
}

private TaskManagerSlot findFreeSlot(
ResourceProfile resourceProfile,
String targetAddress) {
// 1. 优先亲和性：查找与 targetAddress 同节点的 FREE Slot
if (targetAddress != null) {
for (TaskManagerSlot slot : taskManagerSlots.values()) {
if (slot.getState() == SlotState.FREE
&& slot.getTaskManagerLocation().getHostname().equals(targetAddress)
&& slot.getResourceProfile().isMatching(resourceProfile)) {
return slot;
}
}
}
// 2. 资源匹配：查找满足资源需求的 FREE Slot
for (TaskManagerSlot slot : taskManagerSlots.values()) {
if (slot.getState() == SlotState.FREE
&& slot.getResourceProfile().isMatching(resourceProfile)) {
return slot;
}
}
return null; // 无可用 Slot
}

@Override
public void notifyAllocationComplete(AllocationID allocationId, SlotID slotId) {
// 1. 查找对应的 PendingSlotRequest
PendingSlotRequest pendingRequest = pendingSlotRequests.remove(allocationId);
if (pendingRequest == null) {
return;
}

// 2. 查找对应的 TaskManagerSlot
TaskManagerSlot slot = 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 状态管理

public class TaskManagerSlot {
private final SlotID slotId;
private final ResourceID taskManagerId;
private final TaskManagerLocation taskManagerLocation;
private final ResourceProfile resourceProfile;
private SlotState state;
private AllocationID allocationId;
private JobID jobId;

public TaskManagerSlot(
SlotID slotId,
ResourceID taskManagerId,
TaskManagerLocation taskManagerLocation,
ResourceProfile resourceProfile) {
this.slotId = slotId;
this.taskManagerId = taskManagerId;
this.taskManagerLocation = taskManagerLocation;
this.resourceProfile = resourceProfile;
this.state = SlotState.FREE;
}

public void assignPendingAllocation(AllocationID allocationId, JobID jobId) {
// 前置条件检查：当前状态必须是 FREE
Preconditions.checkState(state == SlotState.FREE,
"Cannot assign pending allocation to slot in state %s", state);
this.state = SlotState.PENDING;
this.allocationId = allocationId;
this.jobId = jobId;
}

public void markAllocated() {
// 前置条件检查：当前状态必须是 PENDING
Preconditions.checkState(state == SlotState.PENDING,
"Cannot mark slot as allocated in state %s", state);
this.state = SlotState.ALLOCATED;
}

public void freeSlot() {
// 释放 Slot，回到 FREE 状态
this.state = SlotState.FREE;
this.allocationId = null;
this.jobId = null;
}

public SlotState getState() { return state; }
public SlotID getSlotId() { return slotId; }
public ResourceID getTaskManagerId() { return taskManagerId; }
public TaskManagerLocation getTaskManagerLocation() { return taskManagerLocation; }
public ResourceProfile getResourceProfile() { return resourceProfile; }
}

public enum SlotState {
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 等待请求

public class PendingSlotRequest {
private final AllocationID allocationId;
private final JobID jobId;
private final ResourceProfile resourceProfile;
private final SlotID assignedSlot;
private final long creationTime;

public PendingSlotRequest(
AllocationID allocationId,
JobID jobId,
ResourceProfile resourceProfile,
SlotID assignedSlot) {
this.allocationId = allocationId;
this.jobId = jobId;
this.resourceProfile = resourceProfile;
this.assignedSlot = assignedSlot;
this.creationTime = System.currentTimeMillis();
}

public boolean isTimeout(long timeout) {
return System.currentTimeMillis() – creationTime > timeout;
}

public AllocationID getAllocationId() { return allocationId; }
public JobID getJobId() { return jobId; }
public ResourceProfile getResourceProfile() { return resourceProfile; }
public SlotID getAssignedSlot() { return assignedSlot; }
}

PendingSlotRequest 封装了等待中的 Slot 请求，包含 allocationId、jobId、resourceProfile、assignedSlot（已分配的 Slot，无空闲时为 null）等请求信息。creationTime 记录请求创建时间，用于超时检测。isTimeout 方法判断请求是否超时，超过 request-timeout（默认5分钟）未分配则取消请求。

3.4 SlotStatusSyncer 状态同步

public class SlotStatusSyncer {
private final ResourceManagerGateway resourceManagerGateway;
private final Map<SlotID, SlotState> slotStatusCache;

public void notifySlotStatusChange(SlotID slotId, SlotState newState) {
// 1. 更新状态缓存
SlotState oldState = slotStatusCache.put(slotId, newState);

// 2. 状态变化时通知 ResourceManager
if (oldState != newState) {
resourceManagerGateway.notifySlotStatus(
slotId,
newState,
System.currentTimeMillis());
}
}

public void syncAllSlotStatus() {
// 全量同步所有 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 处理

无 FREE Slot 时，SlotManager 创建 PendingSlotRequest 等待，同时通知 ResourceManager 通过 ResourceManagerDriver 申请新容器。新 TaskManager 注册后，新 Slot 可用，SlotManager 分配给等待的 PendingSlotRequest。超过 request-timeout（默认5分钟）未分配则取消请求，通知 SlotPool 分配失败。

无空闲处理流程：

超时检查机制

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

• 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 区域。

六、常见问题与最佳实践

常见问题

最佳实践

七、总结

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 分配效率和作业稳定性。