本篇将对 Yarn 调度器中的资源抢占方式进行探究。分析当集群资源不足时，占用量资源少的队列，是如何从其他队列中抢夺资源的。我们将深入源码，一步步分析抢夺资源的具体逻辑。
一、简介
在资源调度器中，以 CapacityScheduler 为例（Fair 类似），每个队列可设置一个最小资源量和最大资源量。其中，最小资源量是资源紧缺情况下每个队列需保证的资源量，而最大资源量则是极端情况下队列也不能超过的资源使用量。
资源抢占发生的原因，是为了提高资源利用率，资源调度器（包括 Capacity Scheduler 和 Fair Scheduler）会将负载较轻的队列的资源暂时分配给负载重的队列。
仅当负载较轻队列突然收到新提交的应用程序时，调度器才进一步将本属于该队列的资源归还给它。
但由于此时资源可能正被其他队列使用，因此调度器必须等待其他队列释放资源后，才能将这些资源“物归原主”，为了防止应用程序等待时间过长，RM 在等待一段时间后强制回收。
开启容器抢占需要配置的参数 yarn-site.xml：
yarn.resourcemanager.scheduler.monitor.enable
yarn.resourcemanager.scheduler.monitor.policies
二、抢占具体逻辑
这里我们主要分析如何选出待抢占容器这一过程。
整理流程如下图所示：
接下来我们深入源码，看看具体的逻辑：
首先 ResourceManager 通过 ResourceManager#createPolicyMonitors 方法创建资源抢占服务：
protected void createPolicyMonitors() {
// 只有 capacity scheduler 实现了 PreemptableResourceScheduler 接口，fair 是如何实现资源抢占的？
if (scheduler instanceof PreemptableResourceScheduler
&& conf.getBoolean(YarnConfiguration.RM_SCHEDULER_ENABLE_MONITORS,
YarnConfiguration.DEFAULT_RM_SCHEDULER_ENABLE_MONITORS)) {
LOG.info("Loading policy monitors");
// 是否配置了 scheduler.monitor.policies
// 默认值是 ProportionalCapacityPreemptionPolicy？ 代码中没看到默认值，但是 yarn-site.xml doc 中有默认值
List<SchedulingEditPolicy> policies = conf.getInstances(
YarnConfiguration.RM_SCHEDULER_MONITOR_POLICIES,
SchedulingEditPolicy.class);
if (policies.size() > 0) {
for (SchedulingEditPolicy policy : policies) {
LOG.info("LOADING SchedulingEditPolicy:" + policy.getPolicyName());
// periodically check whether we need to take action to guarantee
// constraints
// 此处创建了资源抢占服务类。
// 当此服务启动时，会启动一个线程每隔 PREEMPTION_MONITORING_INTERVAL（默认 3s）调用一次
// ProportionalCapacityPreemptionPolicy 类中的 editSchedule方法，
// 【重点】在此方法中实现了具体的资源抢占逻辑。
SchedulingMonitor mon = new SchedulingMonitor(rmContext, policy);
addService(mon);
}
资源抢占服务会启动一个线程每隔 3 秒钟调用配置的抢占规则，这里以 ProportionalCapacityPreemptionPolicy（比例容量抢占规则）为例介绍其中的抢占具体逻辑（editSchedule 方法）：
// ProportionalCapacityPreemptionPolicy#editSchedule
public void editSchedule() {
updateConfigIfNeeded();
long startTs = clock.getTime();
CSQueue root = scheduler.getRootQueue();
// 获取集群当前资源快照
Resource clusterResources = Resources.clone(scheduler.getClusterResource());
// 具体的资源抢占逻辑
containerBasedPreemptOrKill(root, clusterResources);
if (LOG.isDebugEnabled()) {
LOG.debug("Total time used=" + (clock.getTime() - startTs) + " ms.");
}
}
editSchedule 方法很简单，逻辑都被封装到 containerBasedPreemptOrKill() 方法中，我们继续深入。
其中主要分三步：
- 生成资源快照
- 根据规则找出各队列待抢占的容器（重点）
- 执行容器资源抢占 或 kill超时未自动停止的容器
// 仅保留重要逻辑
private void containerBasedPreemptOrKill(CSQueue root,
Resource clusterResources) {
// ------------ 第一步 ------------ （生成资源快照）
// extract a summary of the queues from scheduler
// 将所有队列信息拷贝到 queueToPartitions - Map<队列名, Map<资源池, 队列详情>>。生成快照，防止队列变化造成计算问题。
for (String partitionToLookAt : allPartitions) {
cloneQueues(root, Resources
.clone(nlm.getResourceByLabel(partitionToLookAt, clusterResources)), partitionToLookAt);
}
// ------------ 第二步 ------------ （找出待抢占的容器）
// compute total preemption allowed
// based on ideal allocation select containers to be preemptionCandidates from each queue and each application
// candidatesSelectionPolicies 默认会放入 FifoCandidate