Worker根据Topology的定义及分配到自身的任务情况，计算出发出的消息被那些Task接收，由于Worker上分配的任务可能被调整，因此Worker需要定时的更新这些连接信息。
ZMQ连接信息更新
函数：mk-refresh-connections
方法原型：
| 1 | (defn mk-refresh-connection [worker]) |
方法说明：
- 调用worker-outbound-tasks返回从worker参数接收数据的TaskId集合（outbound-tasks）.
- 定义this函数及相应的回调函数,并将自己注册到定时器:refresh-connections-timer中。
- 回调函数带有callback参数用于监听Zookeeper对应节点变化并同步。
- 调用storm-cluster-state的assignment函数获取与storm-id对应的Topology的任务分配。
- 调用:executor->node+port存储Executor到node+port的映射关系，再调用to-task->node+port根据Executor中的TaskId集合构建TaskId到node+port的映射关系，利用select-keys函数和outbound-tasks集合进行过滤，得到从该Worker接收消息的TaskId到node+port的映射关系，最后调用map-val和endpoint->string函数获取node+port的字符串表示，最终结果为一个TaskId到node+port的哈希表（my-assignment）.
- 对my-assingmnet进行过滤，移除所有存在该worker的TaskId结果保存到needed-assignment。
- 从needed-assignment中获取目标Worker集合（node_port）存储到needed-connections,从needed-assignment中获取目标节点上的TaskId集合存储到needed-tasks中。
- 获取worker节点中缓存的从node+port到ZMQ Socket的哈希表的所有键列表，存储到current-connections中。
- 判断哪些连接需要新建，哪些连接可以关闭，分别保存到new-connections和remove-connections中。
- 调用msg/connect方法根据new-connections中的node+poer创建新的连接，并更新到cached-node+port。
- 将cached-task->node+port更新为my-assignment。
- 调用需要删除的Socket的close方法，将这些Socket从：cached-node+port中移除。
应用过程
| 1 | ：refresh-connections-timer(mk-halting-timer) |
| 2 | refresh-connections (mk-refresh-connections worker) |
| 3 | _(refresh-connections nil) |
| 4 | (schedule-recurring (:refresh-connections-timer worker) 0 (conf TASK-REGRESH-POLL-SECS) refresh-connections). |
代码说明：
- 调用mk-halting-timer创建计时器，该代码在创建Worker数据时调用。
- 创建一个用于更新连接的函数，然后立即执行refresh-connections函数更新ZMQ,然后不断执行该函数，执行间隔为TASK-REFRESH-SECS默认为10秒，在mk-worker函数中调用.
从Zookeeper中获取Topology活跃情况
refresh-storm-active函数获取Topology的状态信息.
方法原型:
| 1 | (defn refresh-storm-active ([worker])…) |
方法说明：
- 提供匿名函数做为callback参数，在相应Zookeeper节点变化时回调。
- 通过：refresh-active-timer计时器完成定期调用，默认为10秒。
（shcedule-recurring(:refresh-active-timer worker) 0 (conf TASK-FEFRESH-POLL_SECS) (partial refresh-active worker)）. - 调用:storm-cluster-state的storm-base方法获取Topology的基础信息。
- 判断该Topology是否处于活跃状态（:active），并将判断结果存储于:storm-active-atom变量中。
Worker中接收函数
Mk-transfer-local-fn函数用于产生并发送消息到Executor的接收队列，同一Worker内部的Executor之间通过该函数传递消息。
方法原型：
| 1 | (defn mk-transfer-local-fn [worker]) |
方法说明：
- 调用:short-executor-receive-queue-map返回Executor中第一个Task的TaskId到该Executor对应的接收队列的映射关系，保存到short-executor-queue-map变量。
- 调用task->short-executor返回从该Worker中的TaskId到Executor中第一个Task的TaskId的映射关系。
- 定义返回的匿名函数，该函数输入为ZMQ收到的一组消息tuple-batch,按照与消息TaskId对应的Executor中第一个Task的TaskId对消息进行分组，变量grouped对应的键为Executor中第一个Task的TaskId，值为属于该Executor的一组消息。
- 通过short-executor-receive-queue-map得到TaskId与Executor相对应的接收消息队列q。
- 调用disruptor/publish方法将收到的消息发送到队列q.
Woker中的发送函数
Mk-transfer-fn用于Executor的数据发送，分别有以下两种情况：
- 目标TaskId与发送TaskId属于同一个Worker,此时不需要跨进程传输消息，可将消息通过mk-tansfer-local-fn发送至接收端Executor的接收队列。
- 消息的目标TaskId跟发送TaskId属于不同的Worker中，此时则将消息序列化（KryoTupleSerializer）后发送至Worker的发送队列，由Worker（mk-tranfer-tuples-handler）负责将队列的消息通过ZMQ发送出去.
方法原型：
| 1 | defn mk-transfer [worker] |
该方法返回一个参数为系列化器serializer和一组消息的函数。
不同Worker间的通信
Worker中有一个额外的线程对transfer-queue（worker对应的消息发送队列）进行监听，Mk-transfer-tuples-handler用于创建对应的消息处理器。
方法原型：
| 1 | defn mk-transfer-tuples-handler [worker] |
- 调用cached-node+port获取Worker中与目标node+port相对应的ZMQ Socket连接，保存到node+port->socket.
- 调用worker-data的cached-task->node+port获取TaskId到node+port的映射，保存到task->node+port.
- 定义一个clojure-handler,对应的函数定义为fn [packets _ batch-end?],第一个参数为一组消息packets,第二个忽略，第三个为为结束标记。
- 调用msg/seng函数将消息发送出去。
发送监听线程的启动：
| 1 | transfer-tuples （mk-transfer-tuples-handler worker） |
| 2 | transfer-thread (disruptor /consume-loop* (:transfer-queue worker) transfer-tuples) |
时间： 2024-09-28 10:10:37