Twitter Storm源代码分析之Tuple是如何发送的
这篇文章里面我们来看一下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?
( 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。
推荐阅读:
相关推荐
dflyzx 2020-09-17
higheels 2020-07-20
Ericbig 2020-07-19
yinyang00 2020-05-15
lijiuchangxin 2020-05-14
yogoma 2020-05-09
jacktangj 2020-04-19
Yyqingmofeige 2020-03-26
idning 2020-02-14
宿舍 2020-01-31
luenxin 2020-01-22
水痕 2020-01-18
chouliqingke 2020-01-08
secondid 2019-12-18
XCodeRush 2019-12-17
ryuhfxz 2019-11-20
qonsnow 2014-03-07
coulder 2019-11-17