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;
}