Storm Worker中的线程

worker中的每个计时器都对应一个java线程，使用计数器进行心跳保持以及获取元数据更新信息。

创建woker 

mk-worker函数用于创建worker进程，主要工作包括：

•   启动相应计时器
•   创建worker中对应的executor
•   启动接收消息线程

 

worker心跳信息 

do-heartbeat函数用户产生worker的心跳信息，这些信息被写入本地文件系统中。

supervisor会读取这些心跳信息以判断worker的状态，然后决定是否重启worker。

心跳信息包括如下信息：
current-time-secs：当前时间
storm-id：topologyId
executors：worker中包含的executer列表
port：worker对应的端口号

 

worker获取executor  

read-worker-executor函数用来计算分配到该worker的executor，它通过调用storm-cluster-state的assignment-info函数获取所有topology的分配信息，然后利用worker的assignment-id以及port进行过滤，获得摸个worker所有的executor，assignment-id对应于node。

 

下载topology配置以及代码 

在执行一个topology任务是，supervisor将从nimbus下载三个文件： 

•   stormconf.ser topology配置项的序列化文件，执行任务前需要反序列化。

•   stormcode.ser topology的定义文件。

•   stormjar.jar  用户的资源文件以及源码等。

 

executor心跳信息 

do-executor-heartbeats函数用来发送Executor的心跳信道到zookeeper中，该心跳信息保存了Executor中Task的运行统计，Nimbus利用这些心跳信息判断Executor是否处于活跃状态，并且在storm ui上显示这些运行信息。

 

topology状态信息 

worker需要获知其执行的topology的状态，如果状态为deactive，spout应停止向外发送消息。
refresh-storm-active函数用于获取topology的状态信息。

 

worker数据接收 

mk-receive-queue-map函数用于为worker中的每一个executor创建接收队列，并将其存储hash表，key为 executorId，值为Disruptor Queue对象。

executorId实际是一个一元数组[startTaskId,endTaskId]，表示要执行任务的区间。

 

 

worker数据发送  

mk-transfer-fn函数主要用于executor数据发送。 

• 消息的目标taskId与发送taskId属于同一个worker，直接发送至接收端executor的接收队列，避免网络开销。
• 消息的目标taskId与发送taskId不属于同一个worker，将消息通过netty发送出去。

 

关闭worker 

•  关闭消息接收线程
•  关闭worker中所有executor线程
•  关闭消息发送队列和线程
•   关闭所有计时器线程
•   从zookeeper中清楚该worker的心跳信息
•   断开zookeeper的链接

worker中线程以及通信关系图