JStorm与Storm源码分析(一)--nimbus-data
2017-07-25 21:49
295 查看
Nimbus里定义了一些共享数据结构,比如nimbus-data.
nimbus-data结构里定义了很多公用的数据,请看下面代码:
上述代码中:
submitted-count表示当前已经提交的Topology的数目;
定义了storm-cluster-state对象,该对象可用于将数据存储到ZooKeeper中以及从ZooKeeper读取数据;
submit-lock表示一个锁对象;
定义了downloaders和uploaders缓存;
注:当用户提交Topology时,系统会创建一个上传流并将其放入uploaders缓存中;当Supervisor从Nimbus下载Topology的jar时,系统会创建一个下载流并将其放入downloaders缓存中。任何一种操作完成时,其所对应的上传或下载流就会被关闭,且流所传递的内容也会被从缓存中移除。
uptime定义了当前Nimbus的启动时间;
定义了一个validator,它可用于对Topology进行检测验证.目前使用的是DefaultValidator.
定义了一个timer计时器,并给出了当计数器处理失败时需要调用的方法。
定义了Nimbus所使用的调度器scheduler.
下面我们看一下JStorm中是如何描述nimbus-data的.
在JStorm中定义了一个NimbusData类数据结构来存放数据.NimbusData定义如下:
注:学习李明老师Storm源码分析的笔记。
欢迎关注下面二维码进行技术交流:
nimbus-data结构里定义了很多公用的数据,请看下面代码:
1 (defn nimbus-data [conf inimbus] 2 (let [forced-scheduler (.getForcedScheduler inimbus)] 3 {:conf conf 4 :inimbus inimbus 5 :submitted-count (atom 0) 6 :storm-cluster-state (cluster/mk-storm-cluster-state conf) 7 :submit-lock (Object.) 8 :heartbeats-cache (atom {}) 9 :downloaders (file-cache-map conf) 10 :uploaders (file-cache-map conf) 11 :uptime (uptime-computer) 12 :validator (new-instance (conf NIMBUS-TOPOLOGY-VALIDATOR)) 13 :timer (mk-timer :kill-fn (fn [t] 14 (log-error t "Error when processing event") 15 (halt-process! 20 "Error when processing an event") 16 )) 17 :scheduler (mk-scheduler conf inimbus) 18 }))
上述代码中:
submitted-count表示当前已经提交的Topology的数目;
定义了storm-cluster-state对象,该对象可用于将数据存储到ZooKeeper中以及从ZooKeeper读取数据;
submit-lock表示一个锁对象;
定义了downloaders和uploaders缓存;
注:当用户提交Topology时,系统会创建一个上传流并将其放入uploaders缓存中;当Supervisor从Nimbus下载Topology的jar时,系统会创建一个下载流并将其放入downloaders缓存中。任何一种操作完成时,其所对应的上传或下载流就会被关闭,且流所传递的内容也会被从缓存中移除。
uptime定义了当前Nimbus的启动时间;
定义了一个validator,它可用于对Topology进行检测验证.目前使用的是DefaultValidator.
定义了一个timer计时器,并给出了当计数器处理失败时需要调用的方法。
定义了Nimbus所使用的调度器scheduler.
下面我们看一下JStorm中是如何描述nimbus-data的.
在JStorm中定义了一个NimbusData类数据结构来存放数据.NimbusData定义如下:
1 public class NimbusData { 2 private Map<Object, Object> conf; 3 //定义当前已经提交的Topology的数目 4 private AtomicInteger submittedCount; 5 //定义了cluster-state对象,该对象可用于将数据存储到Zookeeper中以及从Zookeeper读取数据 6 private StormClusterState stormClusterState; 7 8 // Map<topologyId, Map<taskid, TkHbCacheTime>> 9 private ConcurrentHashMap<String, Map<Integer, TkHbCacheTime>> taskHeartbeatsCache; 10 11 // TODO two kind of value:Channel/BufferFileInputStream 12 //定义了downloades和uploaders缓存,当用户提交Topology的时候,系统会创建一个上传流放入uploaders缓存中, 13 //而当Supervisor从Nimbus下载Topology的jar包时,系统则会创建一个下载流并将其放入downloaders缓存中。 14 //任何一种操作完成时,其所对应的上传或下载流就会被关闭,且流所传递的内容也会被从缓存中移除。 15 private TimeCacheMap<Object, Object> downloaders; 16 private TimeCacheMap<Object, Object> uploaders; 17 // cache thrift response to avoid scan zk too frequently 18 private NimbusCache cache; 19 20 private int startTime; 21 22 //定义了Nimbus所使用的调度器 23 private final ScheduledExecutorService scheduExec; 24 25 private StatusTransition statusTransition; 26 27 private static final int SCHEDULE_THREAD_NUM = 8; 28 29 private final INimbus inimubs; 30 //本地模式 31 private final boolean localMode; 32 //当前Nimubus的启动时间 33 public int uptime() { 34 return (TimeUtils.current_time_secs() - startTime); 35 }
注:学习李明老师Storm源码分析的笔记。
欢迎关注下面二维码进行技术交流:
![](https://images2017.cnblogs.com/blog/480165/201707/480165-20170725214743951-1314632955.jpg)
相关文章推荐
- JStorm与Storm源码分析(一)--nimbus-data
- JStorm与Storm源码分析(一)--nimbus-data
- Storm源码分析--Nimbus-data
- Storm源码分析--Nimbus-data
- jstorm源码分析: nimbus
- JStorm与Storm源码分析(四)--均衡调度器,EvenScheduler
- JStorm与Storm源码分析(四)--均衡调度器,EvenScheduler
- JStorm与Storm源码分析(六)--收集器 IOutputCollector 、OutputCollector
- JStorm与Storm源码分析(八)--计时器工具-mk-timer
- JStorm与Storm源码分析(六)--收集器 IOutputCollector 、OutputCollector
- [置顶] JStorm与Storm源码分析(七)--BasicBoltExecutor与装饰模式
- JStorm与Storm源码分析(三)--Scheduler,调度器
- jstorm集群配置部署过程全纪录(补充:报错解决):java.lang.RuntimeException: No alive nimbus
- JStorm与Storm源码分析(二)--任务分配,assignmen
- [转]JStorm之Nimbus简介
- JStorm与Storm源码分析(三)--Scheduler,调度器
- JStorm与Storm源码分析(七)--BasicBoltExecutor与装饰模式
- JStorm之Nimbus简介
- JStorm与Storm源码分析(三)--Scheduler,调度器
- JStorm与Storm源码分析(二)--任务分配,assignment