簡(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