Storm系列(九)架构分析之Supervisor-同步Nimbus的事件线程

最新推荐文章于 2021-10-28 22:39:11 发布

jianyuanPC

最新推荐文章于 2021-10-28 22:39:11 发布

阅读量600

点赞数

分类专栏： Storm

本文链接： https://blog.csdn.net/jianyuanPC/article/details/53462494

版权

17 篇文章 1 订阅

订阅专栏

Supervisor由三个线程组成，一个计时器线程和两个事件线程.

相关数据结构图：

同步Nimbus的事件线程

执行流程图：

方法原型：

 
 1  (defn mk-synchronize-supervisor [supervisor sync-processes event-manager process-event-manager])
 

参数说明：

Supervisor:supervisor-data对象。

Sync-processes:mk-supervisor中定义的用于同步当前supervisor启动的Worker的方法。

Event-manager:mk-supervisor中定义的用于运行mk-synchronize-supervisor方法的线程。

Process-event-manager:mk-supervisor中定义的运行sync-processes方法的线程。

方法说明：

获取supervisor中的配置信息，storm-local-cluster,isupervisor和local-state对象.
定义sync-callback方法，将mk-synchronize-supervisor方法再次加入到Nimbus任务的事件现场中执行，该方法用来在进行Zookeeper操作时设置回调方法，当Zookeeper中的内容发送变化时也会调用该方法.
调用read-storm-code-locations方法获取<storm-id,master-code-location>集合保存到storm-code-map.
调用read-downloaded-storm-ids方法获取当前已经下载Topology对应的storm id信息.
调用read-assignments方法获取分配给当前Supervisor的任务信息，返回<poer,LocalAssignment>集合保存给new-assignment.
调用assigned-storm-ids-from-port-assignments获取new-assignment中的storm-id,保存到assigned-storm-ids.
获取local-state中保存的LS-LOCAL-ASSIGNMENTS信息做为existing-assignment,<port,Assignment>.
调用download-storm-code下载当前已经分配给该Topology但还没下载到本地的Topology信息.
计算出属于existing-assignment但不属于new-assignment的端口信息，调用isupervisor的killedWorker方法关闭这些端口。
调用isupervisor的assigned方法设置新分配的端口信息.
更新local-state中保存的LS_LOCAL_ASSIGNMENT信息，设置为new-assignment.
更新supervisor-data中的curr-assignment信息，设置为new-assignment.
移除本地已下载但已不在assigned-storm-ids中的Topology.
将sync-process方法添加到process-event-manager中执行.