JStorm源码分析(三)Nimbus的任务分配过程
JStorm源码分析(三):Nimbus的任务分配过程
JStorm源码分析文件
对于整个JStorm源码分析系列,我将JStorm源码分析的文件放在我的GitHub上JStorm源码剖析,欢迎大家fork和star
1 . 在 T o p o l o g y 的 任 务 提 交 中 ,最 后 调 用 S t o r m S u b m i t t e r 的 submitTopology方 法 将 任 务 提 交 给 nimbus;实际上,在StormSubmitter的submitTopology中,是NimbusClient通过RPC(远程过程调 用),调用Nimbus的submitTopologyWithOpts方法或者submitTopology方法。
2.submitTopology()方法是Nimbus 接口中的一个方法,其实现类有Servic eHandler,在 ServiceHandler的submitTopology()方法实现源码如下:
从 上 面 的 源 码 可 以 看 出 , 调 用 s u b m i t T o p o l o g y ( ) 方 法 , 只 是 多 了 一 步 把 SubmitOptions 参 数 设 置 为 的TopologyInitialStatus.ACTIVE过程,实际上,最终还是调 用submitTopologyWithOpts(name,uploadedJarLocation,jsonConf,topology,options)方法。
3.submitTopologyWithOpts()方法的源码如下:
4.从submitTopologyWithOpts()方法的源码可以看出,nimbus的任务分配做了以下两步。
(1)将bolt或spout的任务信息写入到zookeeper中 // STONE_NOTE 在zookeeper中为每一个bolt或spout生成任务信息/ZK/tasks/topoologyId/xxx setupZkTaskInfo(conf, topologyId, stormClusterState);
(2)将任务信息放入到队列中,makeAssignment(topologyName, topologyId, options.get_initial_status());
(3)调用TopologyAssign.push(assignEvent)方法,放入队列中。
(4)TopologyAs ign中开启一条线程,在其run()方法中,不停的从消息队列中获取消息
(5)在run()方法中,调用doT opologyAssignment()方法,将获取到的任务事件消息进行任务分配。
(6)在mkAssignment()方法中,进行分配
(7)Scheduler调用scheduler.assignTasks (context)的方法,进行集群模式的任务分配
(8)在这个taskScheduler.assign()方法中,进行每个worker中运行的Task的数量分配
(9)在getRestAssignedWorkers中实现每个worker中的task平均分配的原则
(10)最终,在TaskAssignContext的构造函数中,实现对Worker的Task数量进行平均分配的原则
5.Topology任务配置信息生成流程:
(1)setupZkTaskInfo的源码如下:
调 用 mkTaskComponentAssignments方 法 , 获 取 T a s k 之 间 任 务 信 息 , 放 入 到 M a p 中 。
(2)mkTaskComponentAssignments的源码如下:
(3)mkTaskInfo的源码如下:
(4)mkTaskMaker源码如下:
最终将任务配置信息放入stormClusterState.set_task(topologyId,taskToTaskInfo)中。
至此,nimbus的任务分配过程完成!
相关系列文章
JStorm源码分析(六)Supervisor启动Worker的过程
微信公众号
有兴趣的同学可以关注小编哟!