作者:
xumingming | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明
网址: http://xumingming.sinaapp.com/727/twitter-storm-code-analysis-tuple-send-proc/
网址: 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方法:
|
1
2
3
4
5
6
|
； worker.clj
(
defn
mk-transfer-
fn
[
transfer-queue
]
(
fn
[
task ^Tuple tuple
]
(.put ^LinkedBlockingQueue
transfer-queue
[
task tuple
]
)
))
|
这个方法其实只是往一个LinkedBlockingQueue里面放入一条新记录(task-id, tuple)
然后这个queue里面的内容会被下面这段代码处理
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
; 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里面被初始化的，看下面的代码：
|
01
02
03
04
05
06
07
08
09
10
11
12
13
|
; 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里面的:
|
1
2
3
4
5
6
|
(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接口:
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
(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?
(-> 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。