Twitter Storm源代码分析之Tuple是如何发送的

转载：http://xumingming.sinaapp.com/727/twitter-storm-code-analysis-tuple-send-proc/

这篇文章里面我们来看一下Storm里面的tuple到底是如何从一个tuple是怎么从一个bolt到另一个bolt上去的。

首先Bolt在发射一个tuple的时候是调用OutputCollector的emit或者emitDirect方法，
而这两个方法最终调用的是clojure代码里面的mk-transfer-fn方法:

帮助123456 ； worker.clj (defn mk-transfer-fn [transfer-queue]  (fn [task ^Tuple tuple]    (.put ^LinkedBlockingQueue           transfer-queue [task tuple])     ))

这个方法其实只是往一个LinkedBlockingQueue里面放入一条新记录(task-id, tuple)
然后这个queue里面的内容会被下面这段代码处理

帮助010203040506070809101112131415161718192021222324252627 ; worker.clj ; 这里面的这个socket到底是什么东西? (async-loop     (fn [^ArrayList drainer          ^KryoTupleSerializer serializer]      ; 从transfer-queue里面取出一个任务来       ; 这个任务其实就是(task, tuple)       (let [felem (.take transfer-queue)]        (.add drainer felem)         (.drainTo transfer-queue drainer))       (read-locked endpoint-socket-lock         ; 获取从node+port到socket的映射         (let [node+port->socket @node+port->socket               ; 获取从task-id到node+port的映射               task->node+port @task->node+port]          (doseq [[task ^Tuple tuple] drainer]            ; 获取task对应的socket             (let [socket                    (node+port->socket                      (task->node+port task))               ; 序列化这个tuple               ser-tuple (.serialize serializer tuple)]              ; 发送这个tuple               (msg/send socket task ser-tuple)               ))           )) ）

从上面代码可见，tuple最终是被序列化之后由msg/send方法通过socket发送给指定的task的。注意上面代码里面的async-loop表示会创建一个单独的线程来执行这些代码。可以storm会起一个独立线程来专门发送待发送的消息的。

我们来看下这个socket到底是个怎么样的东西。这个socket是在worker.clj里面被初始化的，看下面的代码：

帮助01020304050607080910111213 ; socket（worker.clj) (swap! node+port->socket      merge      (into {}        (dofor           [[node port :as endpoint] new-connections]         [endpoint           (msg/connect              mq-context              ((:node->;host assignment) node)              port)           ]         )))

从上面代码可以看出socket其实是msg/connect创建出来的。那 msg/connect到底在做什么呢？ 这个方法是定义在protocol.clj里面的:

帮助123456 (defprotocol Context   (bind [context virtual-port])   (connect [context host port])   (send-local-task-empty [context virtual-port])   (term [context])   )

这里定义的只是一个接口而已，具体的实现是在zmq.clj里面。zmq是ZeroMQ的缩写, 可见storm的supervisor之间就是利用zeromq来传递tuple的。

zmq.clj里面的ZMQCOntext实现了Context接口:

帮助01020304050607080910111213141516171819202122232425262728293031323334 (deftype ZMQContext [context linger-ms ipc?]  ; 实现Context接口   Context   ; 从给定的virtual-port拉消息   (bind [this virtual-port]    (->; context         (mq/socket mq/pull)         (mqvp/virtual-bind virtual-port)         (ZMQConnection.)         ))   ; 给给定的host,port推送消息(push)   (connect [this host port]    (let [url (if ipc?                 (str "ipc://" port "ipc")                 (str "tcp://" host ":" port))]      (->; context           (mq/socket mq/push)           (mq/set-linger linger-ms)           (mq/connect url)           (ZMQConnection.))))   ; 给本地的virtual-port发送一条空消息   (send-local-task-empty [this virtual-port]    (let [pusher            (->; context                (mq/socket mq/push)                (mqvp/virtual-connect virtual-port))]          (mq/send pusher (mq/barr))           (.close pusher)))   (term [this]    (.term context))   ; 实现ZMQContextQuery接口   ZMQContextQuery   (zmq-context [this]    context))

总结一些Twitter Storm对于tuple的处理/创建过程：

Bolt创建一个tuple。
Worker把tuple, 以及这个tuple要发送的地址(task-id)组成一个对象(task-id, tuple)放进待发送队列(LinkedBlockingQueue).
一个单独的线程(async-loop所创建的线程）会取出发送队列里面的每个tuple来处理
Worker创建从当前task到目的task的zeromq连接。
序列化这个tuple并且通过这个zeromq的连接来发送这个tuple。