JStorm 原始碼解析：拓撲任務的資源分配過程

// 獲取指定 topology 的執行資料
TopologyInfo topologyInfo = this.getTopologyInfo(topologyId);
if (topologyInfo == null) {
throw new TException("Failed to get topology info");
}

// 獲取指定的 worker 數目：${topology.upgrade.worker.num}
int workerNum = ConfigExtension.getUpgradeWorkerNum(serializedConf);
// 獲取指定的元件名稱：${topology.upgrade.component}
String component = ConfigExtension.getUpgradeComponent(serializedConf);
// 獲取指定的 worker 列表：${topology.upgrade.workers}
Set<String> workers = ConfigExtension.getUpgradeWorkers(serializedConf);

// 判定 topology master 是不是使用獨立的 worker
if (!ConfigExtension.isTmSingleWorker(serializedConf, topologyInfo.get_topology().get_numWorkers())) {
throw new TException("Gray upgrade requires that topology master to be a single worker, cannot perform the upgrade!");
}

// 灰度釋出
return this.grayUpgrade(topologyId, uploadedJarLocation, topology, serializedConf, component, workers, workerNum);

// 對當前 topology 配置進行規範化，並附加一些必要的配置
Map<Object, Object> stormConf = NimbusUtils.normalizeConf(conf, serializedConf, topology);
LOG.info("Normalized configuration:" + stormConf);

// 合併叢集配置和拓撲配置
Map<Object, Object> totalStormConf = new HashMap<>(conf);
totalStormConf.putAll(stormConf);

// 確定 topology 中各個元件的並行度，保證不超過當前 topology 允許的最大值
StormTopology normalizedTopology = NimbusUtils.normalizeTopology(stormConf, topology, true);

/*
 * 驗證 topology 的基本結構資訊：
 * 1. 驗證 topologyName，元件 ID 是否合法
 * 2. 驗證是否存在缺失 input 宣告的 spout
 * 3. 驗證 woker 和 acker 數目引數配置
 */
Common.validate_basic(normalizedTopology, totalStormConf, topologyId);

StormClusterState stormClusterState = data.getStormClusterState();

// 建立 /local-dir/nimbus/${topology_id}/xxxx 檔案，並將元資料同步到 ZK
this.setupStormCode(topologyId, uploadedJarLocation, stormConf, normalizedTopology, false);

// wait for blob replication before activate topology
this.waitForDesiredCodeReplication(conf, topologyId);

// generate TaskInfo for every bolt or spout in ZK : /ZK/tasks/topoologyId/xxx
// 為當前 topology 在 ZK 上生成 task 資訊：/tasks/${topology_id}
this.setupZkTaskInfo(conf, topologyId, stormClusterState);

// mkdir topology error directory : taskerrors/${topology_id}
String path = Cluster.taskerror_storm_root(topologyId);
stormClusterState.mkdir(path);

String grayUpgradeBasePath = Cluster.gray_upgrade_base_path(topologyId); // gray_upgrade/${topology_id}
stormClusterState.mkdir(grayUpgradeBasePath);
// gray_upgrade/${topology_id}/upgraded_workers
stormClusterState.mkdir(Cluster.gray_upgrade_upgraded_workers_path(topologyId));
// gray_upgrade/${topology_id}/upgrading_workers
stormClusterState.mkdir(Cluster.gray_upgrade_upgrading_workers_path(topologyId));

// 為當前 topology 執行任務分配
LOG.info("Submit topology {} with conf {}", topologyName, serializedConf);
this.makeAssignment(topologyName, topologyId, options.get_initial_status());

// push start event after startup
double metricsSampleRate = ConfigExtension.getMetricSampleRate(stormConf); // ${topology.metric.sample.rate}，預設是 0.05
StartTopologyEvent.pushEvent(topologyId, metricsSampleRate);

this.notifyTopologyActionListener(topologyName, "submitTopology");

public void setupZkTaskInfo(Map<Object, Object> conf, String topologyId, StormClusterState stormClusterState) throws Exception {
// 為當前 topology 追加系統元件，同時基於並行度建立元件對應的 task 資訊，同一個元件的多個 task 資訊具備連續的 ID
Map<Integer, TaskInfo> taskToTaskInfo = this.mkTaskComponentAssignments(conf, topologyId);

// 獲取 topology master 的 ID (這裡使用的是其對應的 task ID)
int masterId = NimbusUtils.getTopologyMasterId(taskToTaskInfo);
TopologyTaskHbInfo topoTaskHbInfo = new TopologyTaskHbInfo(topologyId, masterId);
data.getTasksHeartbeat().put(topologyId, topoTaskHbInfo);
// 建立 /ZK/taskbeats/${topology_id}，並寫入 topologyId 和 topologyMasterId
stormClusterState.topology_heartbeat(topologyId, topoTaskHbInfo);

if (taskToTaskInfo == null || taskToTaskInfo.size() == 0) {
throw new InvalidTopologyException("Failed to generate TaskIDs map");
}
// key is task id, value is task info
// 記錄 task 資訊到 ZK : /ZK/tasks/${topology_id}
stormClusterState.set_task(topologyId, taskToTaskInfo);
}

public Map<Integer, TaskInfo> mkTaskComponentAssignments(Map<Object, Object> conf, String topologyId)
throws IOException, InvalidTopologyException, KeyNotFoundException {
// 從 blobstore 中獲取當前 topology 的配置資訊
Map<Object, Object> stormConf = StormConfig.read_nimbus_topology_conf(topologyId, data.getBlobStore());
// 從 blobstore 中獲取當前 topology 的 StormTopology 物件
StormTopology rawTopology = StormConfig.read_nimbus_topology_code(topologyId, data.getBlobStore());
// 追加一些系統元件到當前 topology 中
StormTopology topology = Common.system_topology(stormConf, rawTopology);
// 為當前 topology 生成 task 資訊，key 是 taskId
return Common.mkTaskInfo(stormConf, topology, topologyId);
}

public Assignment mkAssignment(TopologyAssignEvent event) throws Exception {
String topologyId = event.getTopologyId();
LOG.info("Determining assignment for " + topologyId);

// 1. 基於配置和當前叢集執行狀態建立 topology 任務分配的上下文資訊
TopologyAssignContext context = this.prepareTopologyAssign(event);

// 2. 依據當前的執行模式基於對應節點負載為當前 topology 中的 task 分配 worker
Set<ResourceWorkerSlot> assignments;
if (!StormConfig.local_mode(nimbusData.getConf())) {
// 叢集模式，獲取模式的排程器
ITopologyScheduler scheduler = schedulers.get(DEFAULT_SCHEDULER_NAME); // DefaultTopologyScheduler
// 為當前 topology 中的 task 分配 worker
assignments = scheduler.assignTasks(context);
} else {
// 本地模式
assignments = mkLocalAssignment(context);
}

// 3. 記錄任務分配資訊到 ZK: assignments/${topology_id}
Assignment assignment = null;
if (assignments != null && assignments.size() > 0) {
// 獲取服務中的 supervisorId 及其 hostname 對映資訊
Map<String, String> nodeHost = getTopologyNodeHost(context.getCluster(), context.getOldAssignment(), assignments);
// 獲取 task 的啟動時間：<taskId, start_second>
Map<Integer, Integer> startTimes = getTaskStartTimes(
context, nimbusData, topologyId, context.getOldAssignment(), assignments);

String codeDir = (String) nimbusData.getConf().get(Config.STORM_LOCAL_DIR);

assignment = new Assignment(codeDir, assignments, nodeHost, startTimes);

//the topology binary changed.
if (event.isScaleTopology()) {
assignment.setAssignmentType(Assignment.AssignmentType.ScaleTopology);
}
StormClusterState stormClusterState = nimbusData.getStormClusterState();
// 寫入 assignment 資訊到 ZK: assignments/${topology_id}
stormClusterState.set_assignment(topologyId, assignment);

// update task heartbeat's start time
NimbusUtils.updateTaskHbStartTime(nimbusData, assignment, topologyId);
NimbusUtils.updateTopologyTaskTimeout(nimbusData, topologyId);

LOG.info("Successfully make assignment for topology id " + topologyId + ": " + assignment);
}
return assignment;
}

public List<ResourceWorkerSlot> getAvailableWorkers(
DefaultTopologyAssignContext context, Set<Integer> needAssign, int allocWorkerNum) {

// 1. 計算需要分配的 worker 數目
int reserveWorkers = context.getReserveWorkerNum(); // 需要保留的 worker 數目
int workersNum = this.getAvailableWorkersNum(context); // 當前叢集總的可用的 worker 數目
if ((workersNum - reserveWorkers) < allocWorkerNum) {
// 沒有足夠的 worker 可以分配：可用 worker 數目 - 保留的 worker 數目 < 需要分配的數目
throw new FailedAssignTopologyException("there's no enough worker. allocWorkerNum="
+ allocWorkerNum + ", availableWorkerNum=" + workersNum + ",reserveWorkerNum=" + reserveWorkers);
}
workersNum = allocWorkerNum;

// 記錄分配到的 worker
List<ResourceWorkerSlot> assignedWorkers = new ArrayList<>();

// 2. 分配 worker

// 2.1 處理使用者自定義分配的情況
// 從 needAssign 中移除已經分配的 task，並記錄分配的 worker 到 assignedWorkers 中
this.getRightWorkers(context, needAssign, assignedWorkers, workersNum,
// 獲取使用者自定義分配 worker slot 資訊，排除狀態為 unstopped 的 worker
this.getUserDefineWorkers(context, ConfigExtension.getUserDefineAssignment(context.getStormConf())));

if (ConfigExtension.isUseOldAssignment(context.getStormConf())) {
// 2.2 如果配置指定要複用舊的分配，則優先從舊的分配中選出合適的 worker
this.getRightWorkers(context, needAssign, assignedWorkers, workersNum, context.getOldWorkers());
} else if (context.getAssignType() == TopologyAssignContext.ASSIGN_TYPE_REBALANCE && !context.isReassign()) {
// 2.3 如果是 rebalance 任務分配型別，且可以複用原來的 worker 則將原來分配的 worker 記錄下來
int cnt = 0;
for (ResourceWorkerSlot worker : context.getOldWorkers()) {
if (cnt < workersNum) {
ResourceWorkerSlot resFreeWorker = new ResourceWorkerSlot();
resFreeWorker.setPort(worker.getPort());
resFreeWorker.setHostname(worker.getHostname());
resFreeWorker.setNodeId(worker.getNodeId());
assignedWorkers.add(resFreeWorker);
cnt++;
} else {
break;
}
}
}

LOG.info("Get workers from user define and old assignments: " + assignedWorkers);

int restWorkerNum = workersNum - assignedWorkers.size(); // 還需要分配的 worker 數目
if (restWorkerNum < 0) {
throw new FailedAssignTopologyException(
"Too many workers are required for user define or old assignments. " +
"workersNum=" + workersNum + ", assignedWorkersNum=" + assignedWorkers.size());
}

// 2.4 對於剩下需要的 worker，直接新增 ResourceWorkerSlot 例項物件
for (int i = 0; i < restWorkerNum; i++) {
assignedWorkers.add(new ResourceWorkerSlot());
}

/*
* 3. 遍歷將 worker 分配給相應的 supervisor
* - 如果 worker 指定了 supervisor，則優先分配給指定 supervisor
* - 依據 supervisor 的負載情況優先選擇負載較低的進行分配
*/
List<SupervisorInfo> isolationSupervisors = this.getIsolationSupervisors(context);
if (isolationSupervisors.size() != 0) {
this.putAllWorkerToSupervisor(assignedWorkers, this.getResAvailSupervisors(isolationSupervisors));
} else {
// 為 worker 分配對應的 supervisor
this.putAllWorkerToSupervisor(assignedWorkers, this.getResAvailSupervisors(context.getCluster()));
}
this.setAllWorkerMemAndCpu(context.getStormConf(), assignedWorkers);
LOG.info("Assigned workers=" + assignedWorkers);
return assignedWorkers;
}

private void putAllWorkerToSupervisor(List<ResourceWorkerSlot> assignedWorkers, List<SupervisorInfo> supervisors) {

// 遍歷處理 worker，如果指定了 supervisor，且 supervisor 存在空閒埠，則將其分配給該 supervisor
for (ResourceWorkerSlot worker : assignedWorkers) {
if (worker.getHostname() != null) {
for (SupervisorInfo supervisor : supervisors) {
// 如果當前 worker 對應的 hostname 是該 supervisor，且 supervisor 存在空閒的 worker
if (NetWorkUtils.equals(supervisor.getHostName(), worker.getHostname())
&& supervisor.getAvailableWorkerPorts().size() > 0) {
/*
* 基於當前 supervisor 資訊更新對應的 worker 資訊：
*1. 保證 worker 對應的埠號是當前 supervisor 空閒的，否則選一個 supervisor 空閒的給 worker
*2. 設定 worker 對應的 nodeId 為當前 supervisor 的 ID
*/
this.putWorkerToSupervisor(supervisor, worker);
break;
}
}
}
}

// 更新 supervisor 列表，移除沒有空閒埠的 supervisor
supervisors = this.getResAvailSupervisors(supervisors);

// 對 supervisor 按照空閒埠數由大到小排序
Collections.sort(supervisors, new Comparator<SupervisorInfo>() {

@Override
public int compare(SupervisorInfo o1, SupervisorInfo o2) {
return -NumberUtils.compare(o1.getAvailableWorkerPorts().size(), o2.getAvailableWorkerPorts().size());
}

});

/*
* 按照 supervisor 的負載對 worker 進行分配：
* 1. 優先選擇負載較低的 supervisor 進分配
* 2. 如果 supervisor 都已經過載但還有未分配的 worker，則從過載 supervisor 優先選擇空閒埠較多的進行分配
*/
this.putWorkerToSupervisor(assignedWorkers, supervisors);
}

public List<ResourceWorkerSlot> assign() {
if (tasks.size() == 0) { // 沒有需要再分配的任務
assignments.addAll(this.getRestAssignedWorkers());
return assignments;
}

// 1. 處理設定了 task.on.differ.node=true 的元件，為其在不同 supervisor 節點上分配 worker
Set<Integer> assignedTasks = this.assignForDifferNodeTask();

// 2. 為剩餘 task 分配 worker，不包含系統元件
tasks.removeAll(assignedTasks);
Map<Integer, String> systemTasks = new HashMap<>();
for (Integer task : tasks) {
String name = context.getTaskToComponent().get(task);
if (Common.isSystemComponent(name)) {
systemTasks.put(task, name);
continue;
}
this.assignForTask(name, task);
}

// 3. 為系統元件 task 分配 worker, e.g. acker, topology master...
for (Entry<Integer, String> entry : systemTasks.entrySet()) {
this.assignForTask(entry.getValue(), entry.getKey());
}

// 記錄所有分配了 task 的 worker 集合
assignments.addAll(this.getRestAssignedWorkers());
return assignments;
}