jstorm源码阅读（2） —— supervisor简介

青旅半醒 2022-01-16 06:29 279阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

# jstorm源码阅读（2）—— supervisor简介 #

--------------------

### 模块功能 ###

*  Heartbeat 模块用于汇报supervisor心跳信息，包括hostname，workerports，current time和during time等信息，被AsyncLoopThread封装后默认每隔60s一次。
 *  SyncContainerHb 向类似Apsara或Yarn的资源调度框架汇报容器心跳
 *  SyncProcessEvent 由SyncSupervisorEvent调用，在run方法中根据传递进来的分配信息和Topology id来kill or start workers。
 *  EventManagerImp 从队列中取出事件并执行，that's all。
 *  SyncSupervisorEvent 内部组件包括SyncProcessEvent,EventManager,StormClusterState,LocalState,Heartbeat。 run方法是其唯一执行入口，执行过程如下：

1.  实例化EventManagerZkPusher

RunnableCallback syncCallback = new EventManagerZkPusher(this, syncSupEventManager);

1.  分别以lcoal-zk-assignment-version和lcoal-zk-assignment为key从LocalState中获取task分配信息的快照。

Map<String, Integer> assignmentVersion = (Map<String, Integer>) localState.get(Common.LS_LOCAL_ZK_ASSIGNMENT_VERSION);
    if (assignmentVersion == null) {
        assignmentVersion = new HashMap<String, Integer>();
    }
    Map<String, Assignment> assignments = (Map<String, Assignment>) localState.get(Common.LS_LOCAl_ZK_ASSIGNMENTS);
    if (assignments == null) {
        assignments = new HashMap<String, Assignment>();
    }

1.  将zk中的任务分配信息同步到内存。
2.  获取本地已经下载了的jar包的topology id.
3.  过滤出需要本supervisor节点执行的task，并保证每个端口唯一分配给一个task。
4.  把过滤出的本节点任务信息以local-assignments为key写入到LocalState中。
5.  从zk中获取所有topology id和其对应的代码存放路径，并下载。
6.  删除掉本地无用的topology代码。
7.  以需要本节点运行的任务信息为参数调用SyncProcessEvent的run方法来启动topology。
8.  设置心跳开关为true。 但是有个疑问是这个run方法是谁触发的呢？我们注意到有这样一行代码：

RunnableCallback syncCallback = new EventManagerZkPusher(this, syncSupEventManager);
    ...
    getAllAssignments(assignmentVersion, assignments, syncCallback);

代码中的syncSupEventManager就是EventManagerImp，而EventManagerZkPusher的回调方法就是把this加入到EventManagerImp的队列中然后在另一个线程中由EventManagerImp完成this.run方法的调用。那么EventManagerZkPusher的回调方法是在哪里触发的呢，我们注意到在获取task的发布信息的时候有把syncCallback作为参数传递进去，注册到StormClusterState中，查看StormZkClusterState的构造函数会发现他向zk中注册了一个全局的回调函数，在触发该回调函数时根据参数的不同来选择具体应该回调哪一个RunnableCallback。

*  EventManagerPusher 构造函数以SyncSupervisorEvent和EventManagerImpl为参数，用来把SyncSupervisorEvent时间放入EventManagerImpl的队列中，使用AsyncLoopThread循环执行。
 *  HttpServer 主要提供日志查看功能

OK，以上就是supervisor内部一些模块的简单介绍，总体来说就是以EventManagerPusher和zk事件来驱动一个多线程事件驱动模型。

转载于:https://my.oschina.net/nalenwind/blog/840381

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/5a38c263143b456a883ba7e6297c4bda.png