[转]JStorm之Supervisor简介

@SuppressWarnings("rawtypes")
public SupervisorManger mkSupervisor(Map conf, MQContext sharedContext)
        throws Exception {
    LOG.info("Starting Supervisor with conf " + conf);
    active = new AtomicBoolean(true);
    /*
     * Step 1: cleanup all files in /storm-local-dir/supervisor/tmp
     */
    String path = StormConfig.supervisorTmpDir(conf);
    FileUtils.cleanDirectory(new File(path));
    /*
     * Step 2: create ZK operation instance
     * StromClusterState
     */
    StormClusterState stormClusterState = Cluster
            .mk_storm_cluster_state(conf);
    /*
     * Step 3, create LocalStat
     * LocalStat is one KV database
     * 4.1 create LocalState instance
     * 4.2 get supervisorId, if no supervisorId, create one
     */
    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 threads = new Vector();
    // Step 5 create HeartBeat
    // every supervisor.heartbeat.frequency.secs, write SupervisorInfo to ZK
    String myHostName = NetWorkUtils.hostname();
    int startTimeStamp = TimeUtils.current_time_secs();
    Heartbeat hb = new Heartbeat(conf, stormClusterState, supervisorId,
            myHostName, startTimeStamp, active);
    hb.update();
    AsyncLoopThread heartbeat = new AsyncLoopThread(hb, false, null,
            Thread.MIN_PRIORITY, true);
    threads.add(heartbeat);
    // Step 6 create and start sync Supervisor thread
    // every supervisor.monitor.frequency.secs second run SyncSupervisor
    EventManager processEventManager = new EventManagerImp(false);
    ConcurrentHashMap workerThreadPids = new ConcurrentHashMap();
    //读取$jstorm-local-dir/supervior/localstate中key=local-assignments的value值，根据该值执行workers的kill/start
    SyncProcessEvent syncProcessEvent = new SyncProcessEvent(supervisorId,
            conf, localState, workerThreadPids, sharedContext);
    EventManager syncSupEventManager = new EventManagerImp(false);
    //通过比较$zkroot/assignments/{topologyid}全量数据和本地STORM-LOCAL-DIR/supervisor/stormdist/{topologyid}:
    //1.从nimbus下载有任务分配到本节点的topology的jar和配置数据
    //2.从本地删除已经失效的topology的jar和配置数据
    SyncSupervisorEvent syncSupervisorEvent = new SyncSupervisorEvent(
            supervisorId, conf, processEventManager, syncSupEventManager,
            stormClusterState, localState, syncProcessEvent);
    int syncFrequence = (Integer) conf
            .get(Config.SUPERVISOR_MONITOR_FREQUENCY_SECS);
    EventManagerPusher syncSupervisorPusher = new EventManagerPusher(
            syncSupEventManager, syncSupervisorEvent, active, syncFrequence);
    AsyncLoopThread syncSupervisorThread = new AsyncLoopThread(
            syncSupervisorPusher);
    threads.add(syncSupervisorThread);
    LOG.info("Starting supervisor with id " + supervisorId + " at host " + myHostName);
    // SupervisorManger which can shutdown all supervisor and workers
    return new SupervisorManger(conf, supervisorId, active, threads,
            syncSupEventManager, processEventManager, stormClusterState,
            workerThreadPids);
}

@SuppressWarnings("unchecked")
public void update() {
    SupervisorInfo sInfo = new SupervisorInfo(
            TimeUtils.current_time_secs(), myHostName,
            (List) conf.get(Config.SUPERVISOR_SLOTS_PORTS),
            (int) (TimeUtils.current_time_secs() - startTime));
    try {
        stormClusterState.supervisor_heartbeat(supervisorId, sInfo);
    } catch (Exception e) {
        LOG.error("Failed to update SupervisorInfo to ZK", e);
 
    }
}

nohup java –server
-Djava.library.path="$JAVA.LIBRARY.PATH"
-Dlogfile.name="$topologyid-worker-$port.log"
-Dlog4j.configuration=jstorm.log4j.properties
-Djstorm.home="$JSTORM_HOME"
-cp $JAVA_CLASSSPATH:$JSTORM_CLASSPATH
com.alipay.dw.jstorm.daemon.worker.Worker
topologyid supervisorid port workerid

@SuppressWarnings("unchecked")
@Override
public void run() {
    LOG.debug("Syncing processes");
    try {
        /**
         * Step 1: get assigned tasks from localstat Map
         */
        //1.从本地文件$jstorm-local-dir/supervisor/localstate里读取key=“local-assignments”数据
        Map localAssignments = null;
        try {
            localAssignments = (Map) localState
                    .get(Common.LS_LOCAL_ASSIGNMENTS);
        } catch (IOException e) {
            LOG.error("Failed to get LOCAL_ASSIGNMENTS from LocalState", e);
            throw e;
        }
        if (localAssignments == null) {
            localAssignments = new HashMap();
        }
        LOG.debug("Assigned tasks: " + localAssignments);
        /**
         * Step 2: get local WorkerStats from local_dir/worker/ids/heartbeat
         * Map
         */
        //2.根据localAssignments与workers的hb比对结果得到workers的状态
        Map localWorkerStats = null;
        try {
            localWorkerStats = getLocalWorkerStats(conf, localState,
                    localAssignments);
        } catch (IOException e) {
            LOG.error("Failed to get Local worker stats");
            throw e;
        }
        LOG.debug("Allocated: " + localWorkerStats);
        /**
         * Step 3: kill Invalid Workers and remove killed worker from
         * localWorkerStats
         */
        //3.根据workers的状态值启动/停用相关worker
        Set keepPorts = killUselessWorkers(localWorkerStats);
        // start new workers
        startNewWorkers(keepPorts, localAssignments);
    } catch (Exception e) {
        LOG.error("Failed Sync Process", e);
        // throw e
    }
}

@Override
public void run() {
    LOG.debug("Synchronizing supervisor");
    try {
        RunnableCallback syncCallback = new EventManagerZkPusher(this,
                syncSupEventManager);
        /**
         * Step 1: get all assignments
         * and register /ZK-dir/assignment and every assignment watch
         *
         */
        //1.从zk获取分配完成的任务集assignments:(topologyid -> Assignment)
        //$zkroot/assignments/{topologyid}
        Map assignments = Cluster.get_all_assignment(
                stormClusterState, syncCallback);
        LOG.debug("Get all assignments " + assignments);
        /**
         * Step 2: get topologyIds list from
         * STORM-LOCAL-DIR/supervisor/stormdist/
         */
        //2.本地已经下载的topology集合$jstorm-local-dir/supervisor/stormdist/{topologyid}
        List downloadedTopologyIds = StormConfig
                .get_supervisor_toplogy_list(conf);
        LOG.debug("Downloaded storm ids: " + downloadedTopologyIds);
        /**
         * Step 3: get  from ZK local node‘s
         * assignment
         */
        //3.从assignments里筛选出分配到当前supervisor的任务集合
        Map localAssignment = getLocalAssign(
                stormClusterState, supervisorId, assignments);
        /**
         * Step 4: writer local assignment to LocalState
         */
        //4.将步骤3得到的结果写本地文件$jstorm-local-dir/supervisor/localstate
        try {
            LOG.debug("Writing local assignment " + localAssignment);
            localState.put(Common.LS_LOCAL_ASSIGNMENTS, localAssignment);
        } catch (IOException e) {
            LOG.error("put LS_LOCAL_ASSIGNMENTS " + localAssignment
                    + " of localState failed");
            throw e;
        }
        // Step 5: download code from ZK
        //5.下载新分配任务的Topology
        Map topologyCodes = getTopologyCodeLocations(assignments);
        downloadTopology(topologyCodes, downloadedTopologyIds);
        /**
         * Step 6: remove any downloaded useless topology
         */
        6.删除过期任务的Topology
        removeUselessTopology(topologyCodes, downloadedTopologyIds);
        /**
         * Step 7: push syncProcesses Event
         */
        processEventManager.add(syncProcesses);
    } catch (Exception e) {
        LOG.error("Failed to Sync Supervisor", e);
        // throw new RuntimeException(e);
    }
}

public WorkerShutdown execute() throws Exception {
	//1. Tuple序列化+发送管道LinkedBlockingQueue
    WorkerTransfer workerTransfer = getSendingTransfer();
    // shutdown task callbacks
    //2. 初始化task线程
    List shutdowntasks = createTasks(workerTransfer);
    workerData.setShutdownTasks(shutdowntasks);
    //3. WorkerVirtualPort：tuple接收dispatcher
    // create virtual port object
    // when worker receives tupls, dispatch targetTask according to task_id
    // conf, supervisorId, topologyId, port, mqContext, taskids
    WorkerVirtualPort virtual_port = new WorkerVirtualPort(workerData);
    Shutdownable virtual_port_shutdown = virtual_port.launch();
    //3. RefreshConnections：维护节点间的连接：创建新连接|维护已建立连接|销毁无用连接
    // refresh connection
    RefreshConnections refreshConn = makeRefreshConnections();
    AsyncLoopThread refreshconn = new AsyncLoopThread(refreshConn);
    // refresh ZK active status
    RefreshActive refreshZkActive = new RefreshActive(workerData);
    AsyncLoopThread refreshzk = new AsyncLoopThread(refreshZkActive);
    //4. WorkerHeartbeatRunable：心跳线程
    // 每次心跳更新本地目录数据 $LOCAL_PATH/workers/{worker-id}/Heartbeats/{worker-id}
    // refresh hearbeat to Local dir
    RunnableCallback heartbeat_fn = new WorkerHeartbeatRunable(workerData);
    AsyncLoopThread hb = new AsyncLoopThread(heartbeat_fn, false, null,
            Thread.NORM_PRIORITY, true);
    //5. DrainerRunable：发送tuple线程
    // transferQueue, nodeportSocket, taskNodeport
    DrainerRunable drainer = new DrainerRunable(workerData);
    AsyncLoopThread dr = new AsyncLoopThread(drainer, false, null,
            Thread.MAX_PRIORITY, true);
    AsyncLoopThread[] threads = { refreshconn, refreshzk, hb, dr };
    //6. 注册主线程退出数据清理hook
    return new WorkerShutdown(workerData, shutdowntasks,
            virtual_port_shutdown, threads);
}