JStorm之NimbusServer启动流程

   NimbusServer相当于hadoop里的JobTracker或yarn里的ResourceManager，在集群中属于首脑地位，负责分发任务，监控集群状态，与supervisor的通信主要通过Zookeeper。nimbus在启动过程中会做以下工作，以保证集群稳定运行：
1、清理无效topology
2、建立zk连接并创建相应znode
3、启动监控线程
4、启动httpserver

启动主函数如下：

public static void main(String[] args) throws Exception {
// 读取配置文件，如果我们在JVM启动参数中包含了storm.conf.file系统属性
//则会直接读取参数指定的文件，否则会找strom.yaml，这个一个yaml格式的文件
//读取出的配置信息最终放入Map中
@SuppressWarnings("rawtypes")
Map config = Utils.readStormConfig();
NimbusServer instance = new NimbusServer();
//处理资源分配的工具
INimbus iNimbus = new DefaultInimbus();
//主要启动流程包含此函数中
instance.launchServer(config, iNimbus);
}

nimbus启动的主方法如下：

private void launchServer(final Map conf, INimbus inimbus) {
LOG.info("Begin to start nimbus with conf " + conf);

try {
// 检测集群配置是本地模式还是集群模式
StormConfig.validate_distributed_mode(conf);
//清理pid目录，由参数storm.local.dir指定，该过程包含如下操作
//1、创建本地进程的pid文件
//2、删除历史pid文件，比如上次进程直接被kill
createPid(conf);
//初始化关闭钩子线程：里面包含各种清理操作，可参考cleanup方法
initShutdownHook();
//初始化ininumbus目录
inimbus.prepare(conf, StormConfig.masterInimbus(conf));
//创建数据容器，存放各种统计信息例如提交此俗、集群状态、启动时间，参考下面NimbusData类成员
data = createNimbusData(conf, inimbus);
//启动fllower线程不断更新zk状态，如果发现自己不是leader则会关闭进程
initFollowerThread(conf);
//启动http服务器
int port = ConfigExtension.getNimbusDeamonHttpserverPort(conf);
hs = new Httpserver(port, conf);
hs.start();
//如果不是运行在yarn类似的容器中，不用关心此方法
initContainerHBThread(conf);

while (!data.isLeader())
Utils.sleep(5000);

//初始换监控线程
initUploadMetricThread(data);
//进行初始化操作，该方法会清理zk中残留的topology信息，初始化TopologyAssign线程负责分配任务
init(conf);
} catch (Throwable e) {
LOG.error("Fail to run nimbus ", e);
} finally {
cleanup();
}

LOG.info("Quit nimbus");
}

NimbusServer的init方法如下：

private void init(Map conf) throws Exception {

NimbusUtils.cleanupCorruptTopologies(data);
//初始化TopologyAssign线程
initTopologyAssign();

initTopologyStatus();
//jar文件清理线程，默认每10分钟会清理一次，清理目录：$LOCAL-DIR/nimbus/inbox
initCleaner(conf);

serviceHandler = new ServiceHandler(data);

if (!data.isLocalMode()) {
//监控各task运行状态，发现dead状态的则会清理掉释放资源
initMonitor(conf);
//启动Thrift server
initThrift(conf);

}
}

最后附上NimbusData的数据结构：

public class NimbusData {

private Map<Object, Object> conf;

private StormClusterState stormClusterState;

// Map<topologyId, Map<taskid, TkHbCacheTime>>
private ConcurrentHashMap<String, Map<Integer, TkHbCacheTime>> taskHeartbeatsCache;

// TODO two kind of value:Channel/BufferFileInputStream
private TimeCacheMap<Object, Object> downloaders;
private TimeCacheMap<Object, Object> uploaders;

private int startTime;

private final ScheduledExecutorService scheduExec;

private AtomicInteger submittedCount;

private Object submitLock = new Object();

private StatusTransition statusTransition;

private static final int SCHEDULE_THREAD_NUM = 8;

private final INimbus inimubs;

private Map<String, Map<String, Map<ThriftResourceType, Integer>>> groupToTopology;

private Map<String, Map<ThriftResourceType, Integer>> groupToResource;

private Map<String, Map<ThriftResourceType, Integer>> groupToUsedResource;

private final boolean localMode;

private volatile boolean isLeader;

...........
}