[转]JStorm之Nimbus简介

@SuppressWarnings("rawtypes")
private void launchServer(Map conf) throws Exception {
    LOG.info("Begin to start nimbus with conf " + conf);
    //1.检查配置文件中是否配置为分布式模式
    StormConfig.validate_distributed_mode(conf);
    //2.注册主线程退出Hook现场清理（关闭线程+清理数据）
    initShutdownHook();
    //3.新建NimbusData数据，记录30s超时上传下载通道Channel/BufferFileInputStream
    data = createNimbusData(conf);
    //4.nimbus本地不存在的stormids数据如果在ZK上存在则删除，其中删除操作包括/zk/{assignments,tasks,storms}相关数据
    NimbusUtils.cleanupCorruptTopologies(data);
    //5.启动Topology分配策略
    initTopologyAssign();
    //6.初始化所有topology的状态为startup
    initTopologyStatus();
    //7.监控所有task的heartbeat,一旦发现taskid失去心跳将其置为needreassign 1次/10s
    initMonitor(conf);
    //8.启动cleaner线程，默认600s扫描一次，默认删除3600s没有读写过的jar文件
    initCleaner(conf);
    //9.初始化ServiceHandler
    serviceHandler = new ServiceHandler(data);
    //10.启动rpc server
    initThrift(conf);
}

public static void submitTopology(
String name,
Map stormConf,
StormTopology topology)
throws AlreadyAliveException, InvalidTopologyException {
  if(!Utils.isValidConf(stormConf)) {
      throw new IllegalArgumentException("Storm conf is not valid.");
  }
  stormConf = new HashMap(stormConf);
  stormConf.putAll(Utils.readCommandLineOpts());
  Map conf = Utils.readStormConfig();
  conf.putAll(stormConf);
  try {
      String serConf = JSONValue.toJSONString(stormConf);
      if(localNimbus!=null) {
          LOG.info("Submitting topology " + name + " in local mode");
          localNimbus.submitTopology(name, null, serConf, topology);
      } else {
          //1.向Nimbus提交jar包
          submitJar(conf);
          NimbusClient client = NimbusClient.getConfiguredClient(conf);
          try {
              LOG.info("Submitting topology " +  name + " in distributed mode with conf "  + serConf);
              //2.提交topology DAG及序列化后的配置信息serconf(json)
              client.getClient().submitTopology(name, submittedJar, serConf, topology);
          } finally {
              client.close();
          }
      }
      LOG.info("Finished submitting topology: " +  name);
  } catch(TException e) {
      throw new RuntimeException(e);
  }
}

@SuppressWarnings("unchecked")
@Override
public void submitTopology(String topologyname, String uploadedJarLocation, String jsonConf, StormTopology topology)
        throws AlreadyAliveException, InvalidTopologyException, TException {
    ……
    try {
    	//1.检测topologyName是否已经存在，如果存在相同名称的topology则拒绝提交
        checkTopologyActive(data, topologyname, false);
    }
    ……
    //2.根据topologyname构造topologyid(=topologyname-$counter-$ctime)
    int counter = data.getSubmittedCount().incrementAndGet();
    String topologyId = topologyname + "-" + counter + "-"
            + TimeUtils.current_time_secs();
    //3.根据输入参数jsonConf重组配置数据
    Map serializedConf = (Map) JStormUtils.from_json(jsonConf);
    if (serializedConf == null) {
        throw new InvalidTopologyException("");
    }
    serializedConf.put(Config.STORM_ID, topologyId);
    Map stormConf;
    try {
        stormConf = NimbusUtils.normalizeConf(conf, serializedConf, topology);
    } catch (Exception e1) {
        throw new TException(errMsg);
    }
    Map totalStormConf = new HashMap(conf);
    totalStormConf.putAll(stormConf);
    StormTopology newtopology = new StormTopology(topology);
    //4.检查topology的合法性,包括componentid检查和spout/bolt不能为空检查
    // this validates the structure of the topology
    Common.validate_basic(newtopology, totalStormConf, topologyId);
    try {
        StormClusterState stormClusterState = data.getStormClusterState();
        //5.在nimbus的本地准备所有topology相关数据
        //包括$storm-local-dir/nimbus/stormdist/topologyid/{tormjar.jar,stormcode.ser,stormconf.ser}
        // create $storm-local-dir/nimbus/topologyId/xxxx files
        setupStormCode(conf, topologyId, uploadedJarLocation, stormConf,
                newtopology);
        //6.向zk写入task信息
        //6.1新建目录$zkroot/taskbeats/topologyid
        //6.2写文件$zkroot/tasks/topologyid/taskid 内容为对应task的taskinfo[内容：componentid]
        // generate TaskInfo for every bolt or spout in ZK
        // $zkroot /tasks/topoologyId/xxx
        setupZkTaskInfo(conf, topologyId, stormClusterState);
        //7.任务分配事件压入待分配队列
        // make assignments for a topology
        TopologyAssignEvent assignEvent = new TopologyAssignEvent();
        assignEvent.setTopologyId(topologyId);
        assignEvent.setScratch(false);
        assignEvent.setTopologyName(topologyname);
        TopologyAssign.push(assignEvent);
    }
    ……
}

public static List sortSlots(
Set allSlots, int needSlotNum) {
    Map> nodeMap = new HashMap>();
    // group by first
    // 按照节点进行组织Map> : nodeid -> ports
    for (NodePort np : allSlots) {
        String node = np.getNode();
        List list = nodeMap.get(node);
        if (list == null) {
            list = new ArrayList();
            nodeMap.put(node, list);
        }
        list.add(np);
    }
 
    //每一个nodeid按照端口的大小进行排序
    for (Entry> entry : nodeMap.entrySet()) {
        List ports = entry.getValue();
        Collections.sort(ports);
    }
 
    //收集所有的workers
    List> splitup = new ArrayList>(nodeMap.values());
 
    //按照节点可用worker数从小到大排序
    //1.assignTasks-Map supInfos
    //2.availSlots : splitup/List>
    Collections.sort(splitup, new Comparator> () {
        public int compare(List o1, List o2) {
            return o1.size() - o2.size();
        }
    });
 
    /*
     * splitup目前的状态(A-F表示节点，1-h表示端口)
     * |A| |B| |C| |D| |E| |F|
     *--|---|---|---|---|---|--
     * |1| |2| |3| |4| |5| |6|
     *     |7| |8| |9| |0| |a|
     *         |b| |c| |d| |e|
     *                 |f| |g|
     *                     |h|
     * 经过interleave_all收集到的sortedFreeSlots为：
     * 1-2-3-4-5-6-7-8-9-0-a-b-c-d-e-f-g-h
     */
    List sortedFreeSlots = JStormUtils.interleave_all(splitup);
 
    //比较sortedFreeSlots.size和needSlotNum的大小分配workers
    if (sortedFreeSlots.size()  needSlotNum
    return sortedFreeSlots.subList(0, needSlotNum);
}

@Override
public void run() {
    //1.获取jstorm对zk的操作接口
    StormClusterState clusterState = data.getStormClusterState();
    try {
        // Attetion, here don‘t check /ZK-dir/taskbeats to
        // get active topology list
        //2.通过$zkroot/assignments/获取所有需要检查active topology
        List active_topologys = clusterState.assignments(null);
 
        if (active_topologys == null) {
            LOG.info("Failed to get active topologies");
            return;
        }
 
        for (String topologyid : active_topologys) {
            LOG.debug("Check tasks " + topologyid);
            // Attention, here don‘t check /ZK-dir/taskbeats/topologyid to
            // get task ids
            //3.通过$zkroot/tasks/topologyid获取组成topology的所有tasks
            List taskIds = clusterState.task_ids(topologyid);
            if (taskIds == null) {
                LOG.info("Failed to get task ids of " + topologyid);
                continue;
            }
 
            boolean needReassign = false;
            for (Integer task : taskIds) {
                //4.检查task是否为Dead状态，主要是task心跳是否超时
                boolean isTaskDead = NimbusUtils.isTaskDead(data, topologyid, task);
                if (isTaskDead == true) {
                    needReassign = true;
                    break;
                }
            }
            if (needReassign == true) {
                //5.如果Topology里有Dead状态的Task则topology状态置为monitor等待任务分配线程重新分配
                NimbusUtils.transition(data, topologyid, false, StatusType.monitor);
            }
        }
    } catch (Exception e) {
        // TODO Auto-generated catch block
        LOG.error(e.getCause(), e);
    }
}