JStorm之Supervisor启动流程

Supervisor中文翻译是监督者，意思简单明了，就是对资源进行监控，其实主要是woker资源。该组件所做的事情概括如下：

1、每隔一段时间发送心跳证明自己还活着

2、下载新的topology

3、释放无效的woker

4、分配新的任务

该组件主要包含：心跳线程、supervisor事件接受线程、处理线程，一旦事件接受到则会进入任务分配环节，主要逻辑代码如下：

public static void main(String[] args) {
	Supervisor instance = new Supervisor();
	//主要初始化操作方法
	instance.run();
}

public void run() {
	SupervisorManger supervisorManager = null;
	try {
		//读取配置文件，和nimbus一样，不多解释
		Map<Object, Object> conf = Utils.readStormConfig();
		//获得集群模式
		StormConfig.validate_distributed_mode(conf);
		//创建pid文件
		createPid(conf);
		//supervisor会在该方法中启动
		supervisorManager = mkSupervisor(conf, null);
		JStormUtils.redirectOutput("/dev/null");

	} catch (Exception e) {
		LOG.error("Failed to start supervisor\n", e);
		System.exit(1);
	}
	while (supervisorManager.isFinishShutdown() == false) {
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {

		}
	}
}

下面看组件启动的主方法

public SupervisorManger mkSupervisor(Map conf, IContext sharedContext)
			throws Exception {
		LOG.info("Starting Supervisor with conf " + conf);
		active = new AtomicBoolean(true);
		/**
		 * Step 1: 清理临时文件：$storm.local.dir/supervisor/tmp
		 */
		String path = StormConfig.supervisorTmpDir(conf);
		FileUtils.cleanDirectory(new File(path));
		/*
		 * Step 2: 创建zk连接，并建立一系列目录：[/assignments, /tasks, /topology, /supervisors, /taskbeats, /taskerrors, /monitor]
		 *
		 */
		StormClusterState stormClusterState = Cluster
				.mk_storm_cluster_state(conf);
		/*
		 * Step 3  创建$storm.local.dirsupervisor/localstate目录
		 *         创建以时间戳为名的版本文件：1421217778765、1421217778765.version并将supervisorid序列化到文件内
		 *         删除旧的版本文件，保留最近4个
		 */
		LocalState localState = StormConfig.supervisorState(conf);
		String supervisorId = (String) localState.get(Common.LS_ID);
		if (supervisorId == null) {
			supervisorId = UUID.randomUUID().toString();
			localState.put(Common.LS_ID, supervisorId);
		}
		Vector<SmartThread> threads = new Vector<SmartThread>();
		// Step 5 create HeartBeat
		/*
		* 创建heatbeat线程，并每隔supervisor.heartbeat.frequency.secs秒发送一次心跳，更新zk里的znode
		* znode节点为：/supervisors/supervisor-id (0add54ac-2c23-49bc-aaee-05b3cb9fef00)
		* 会更新该节点的如下信息：SupervisorInfo[hostName=rt2l02046.tbc,
		*					 										supervisorId=0add54ac-2c23-49bc-aaee-05b3cb9fef00,
		*					 										timeSecs=1421219320,
		*					 										uptimeSecs=908,
		*					 										workerPorts=[6801, 6800, 6803, 6802]]
		*
		*/
		Heartbeat hb = new Heartbeat(conf, stormClusterState, supervisorId,
				active);
		hb.update();
		AsyncLoopThread heartbeat = new AsyncLoopThread(hb, false, null,
				Thread.MIN_PRIORITY, true);
		threads.add(heartbeat);

		// Sync heartbeat to Apsara Container
		AsyncLoopThread syncContainerHbThread = SyncContainerHb.mkSupervisorInstance(conf);
		if (syncContainerHbThread != null) {
		    threads.add(syncContainerHbThread);
		}
		// Step 6 create and start sync Supervisor thread
		// every supervisor.monitor.frequency.secs second run SyncSupervisor
		/*
		*  创建两个同步线程，SyncSupervisorEvent和SyncProcessEvent分别用来接收事件和处理事件，
		*  与topology关系比较密切，在topology提交的时候再详细介绍
		*/
		EventManager processEventManager = new EventManagerImp(false);
		ConcurrentHashMap<String, String> workerThreadPids = new ConcurrentHashMap<String, String>();
		SyncProcessEvent syncProcessEvent = new SyncProcessEvent(supervisorId,
				conf, localState, workerThreadPids, sharedContext);

		EventManager syncSupEventManager = new EventManagerImp(false);
		SyncSupervisorEvent syncSupervisorEvent = new SyncSupervisorEvent(
				supervisorId, conf, processEventManager, syncSupEventManager,
				stormClusterState, localState, syncProcessEvent);

		int syncFrequence = JStormUtils.parseInt(conf
				.get(Config.SUPERVISOR_MONITOR_FREQUENCY_SECS));
		EventManagerPusher syncSupervisorPusher = new EventManagerPusher(
				syncSupEventManager, syncSupervisorEvent, active, syncFrequence);
		AsyncLoopThread syncSupervisorThread = new AsyncLoopThread(
				syncSupervisorPusher);
		threads.add(syncSupervisorThread);

		//Step 7 start httpserver
		int port = ConfigExtension.getSupervisorDeamonHttpserverPort(conf);
		Httpserver httpserver = new Httpserver(port, conf);
		httpserver.start();

		//Step 8 上传监控信息
		MetricSendClient client;
		if (ConfigExtension.isAlimonitorMetricsPost(conf)) {
			client = new AlimonitorClient(AlimonitorClient.DEFAUT_ADDR,
					AlimonitorClient.DEFAULT_PORT, true);
		} else {
		    client = new MetricSendClient();
		}
		UploadSupervMetric uploadMetric = new UploadSupervMetric(conf, stormClusterState,
				supervisorId, active, 60, client);
		AsyncLoopThread uploadMetricThread = new AsyncLoopThread(uploadMetric);
		threads.add(uploadMetricThread);

		// SupervisorManger which can shutdown all supervisor and workers
		return new SupervisorManger(conf, supervisorId, active, threads,
				syncSupEventManager, processEventManager, httpserver,
				stormClusterState, workerThreadPids);
	}