簡(jiǎn)介

源碼解析

Capacity 調度器的核心類(lèi)是CapacityScheduler。在初始化CapacityScheduler的時(shí)候，在構造函數initAsyncSchedulingProperties，里面會(huì )初始化調度器相關(guān)。
核心類(lèi)是AsyncSchedulingConfiguration，主要內容總結為：初始化異步調度器線(xiàn)程AsyncScheduleThread，可以初始化多個(gè)，調度支持多線(xiàn)程。

AsyncScheduleThread繼承自Thread，核心是循環(huán)調度，調度的核心函數為schedule。

schedule函數

一般情況下，滿(mǎn)足下面條件的節點(diǎn)不會(huì )被分配資源：

• 心跳超時(shí)的節點(diǎn)，心跳超時(shí)的節點(diǎn)一般認為是可能已經(jīng)dead了。為了可靠性考慮，不給此類(lèi)節點(diǎn)分配Container。
• 當前節點(diǎn)的狀態(tài)不為RUNNING狀態(tài)，不為RUNNING狀態(tài)的節點(diǎn)是異常的，不能分配節點(diǎn)。

上述判斷的核心實(shí)現函數為shouldSkipNodeSchedule。

資源分配方式

資源分配方式分為：

• 按照節點(diǎn)分配資源
• 按照標簽進(jìn)行分配

按照節點(diǎn)分配資源

• 隨機產(chǎn)生一個(gè)隨機數，范圍是0 ~ allNode.size。
• 優(yōu)先從下標為[start, end)的節點(diǎn)中分配資源。
• 再次從下標為[0, start)的節點(diǎn)中分配資源。

代碼主要流程如下：

int start = random.nextInt(nodeSize);
boolean printSkippedNodeLogging = isPrintSkippedNodeLogging(cs);

// Allocate containers of node [start, end)
for (FiCaSchedulerNode node : nodes) {
  if (current++ >= start) {
    if (shouldSkipNodeSchedule(node, cs, printSkippedNodeLogging)) {
      continue;
    }
    cs.allocateContainersToNode(node.getNodeID(), false);
  }
}

current = 0;

// Allocate containers of node [0, start)
for (FiCaSchedulerNode node : nodes) {
  if (current++ > start) {
    break;
  }
  if (shouldSkipNodeSchedule(node, cs, printSkippedNodeLogging)) {
    continue;
  }
  cs.allocateContainersToNode(node.getNodeID(), false);
}

按照標簽進(jìn)行分配

• 隨機產(chǎn)生一個(gè)隨機數，范圍是0 ~ partitions.size。
• 優(yōu)先從下標為[start, end)的標簽中分配資源。
• 再次從下標為[0, start)的標簽中分配資源。

int partitionSize = partitions.size();
// First randomize the start point
int start = random.nextInt(partitionSize);
// Allocate containers of partition [start, end)
for (String partition : partitions) {
  if (current++ >= start) {
    CandidateNodeSet<FiCaSchedulerNode> candidates =
            cs.getCandidateNodeSet(partition);
    if (candidates == null) {
      continue;
    }
    cs.allocateContainersToNode(candidates, false);
  }
}

current = 0;

// Allocate containers of partition [0, start)
for (String partition : partitions) {
  if (current++ > start) {
    break;
  }
  CandidateNodeSet<FiCaSchedulerNode> candidates =
          cs.getCandidateNodeSet(partition);
  if (candidates == null) {
    continue;
  }
  cs.allocateContainersToNode(candidates, false);
}

資源分配具體實(shí)現

資源分配的核心實(shí)現函數為allocateContainersToNode。首先檢查當前節點(diǎn)是否存在運行時(shí)預留的資源，優(yōu)先處理運行時(shí)預留資源。

運行時(shí)預留

資源分配

對于可用資源和可kill的資源加和小于最小資源的時(shí)候，不會(huì )再進(jìn)行資源分配或者資源預留了，因為資源肯定是不足的。

資源存在的場(chǎng)景需要進(jìn)行資源分配或者資源預留。核心實(shí)現函數為allocateOrReserveNewContainers。優(yōu)先嘗試從沒(méi)有標簽的節點(diǎn)分配資源。再沒(méi)有分配到資源之后，最后嘗試按照資源標簽進(jìn)行分配。

從沒(méi)有資源標簽的節點(diǎn)分配的函數入口如下，資源分配都是從根隊列開(kāi)始分配的。

CSAssignment assignment = getRootQueue().assignContainers(
    getClusterResource(), candidates, new ResourceLimits(labelManager
        .getResourceByLabel(candidates.getPartition(),
            getClusterResource())),
    SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
assignment.setSchedulingMode(SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
submitResourceCommitRequest(getClusterResource(), assignment);

對于包含資源標簽的節點(diǎn)分配資源實(shí)現如下：

assignment = getRootQueue().assignContainers(getClusterResource(),
    candidates,
    // TODO, now we only consider limits for parent for non-labeled
    // resources, should consider labeled resources as well.
    new ResourceLimits(labelManager
        .getResourceByLabel(RMNodeLabelsManager.NO_LABEL,
            getClusterResource())),
    SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY);
assignment.setSchedulingMode(SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY);
submitResourceCommitRequest(getClusterResource(), assignment);

assignContainers

根隊列的實(shí)現類(lèi)為AbstractParentQueue.java。低版本的實(shí)現類(lèi)為ParentQueue.java