Storm-源码分析-Topology Submit-Executor-mk-threads

1.SpoutThread

(defmethodmk-threads:spout[executor-datatask-datas]
(let[{:keys[storm-confcomponent-idworker-contexttransfer-fnreport-errorsampleropen-or-prepare-was-called?]}executor-data
;;1.1定义pending

^ISpoutWaitStrategyspout-wait-strategy(init-spout-wait-strategystorm-conf)
max-spout-pending(executor-max-spout-pendingstorm-conf(counttask-datas))
^Integermax-spout-pending(ifmax-spout-pending(intmax-spout-pending))
last-active(atomfalse)
spouts(ArrayList.(map:object(valstask-datas)))
rand(Random.(Utils/secureRandomLong))
pending(RotatingMap.
2;;microoptimizeforperformanceof.sizemethod
(reifyRotatingMap$ExpiredCallback
(expire[thismsg-id[task-idspout-idtuple-infostart-time-ms]]
(let[time-delta(ifstart-time-ms(time-delta-msstart-time-ms))];;start-time-ms是取样赋值的,一般为null,只有有start-time-ms,才会产生time-delta
(fail-spout-msgexecutor-data(gettask-datastask-id)spout-idtuple-infotime-delta)
))))

;;1.2定义tuple-action-fn
tuple-action-fn(fn[task-id^TupleImpltuple]
(let[stream-id(.getSourceStreamIdtuple)]
(condp=stream-id
Constants/SYSTEM_TICK_STREAM_ID(.rotatepending)
Constants/METRICS_TICK_STREAM_ID(metrics-tickexecutor-datatask-datastuple)
(let[id(.getValuetuple0);;tuplevalues,values[0]为id
[stored-task-idspout-idtuple-finished-infostart-time-ms](.removependingid)];;从pending中删除tuple，重要！
(whenspout-id
(when-not(=stored-task-idtask-id)
(throw-runtime"Fatalerror,mismatchedtaskids:"task-id""stored-task-id))
(let[time-delta(ifstart-time-ms(time-delta-msstart-time-ms))]
(condp=stream-id
ACKER-ACK-STREAM-ID(ack-spout-msgexecutor-data(gettask-datastask-id);;ack
spout-idtuple-finished-infotime-delta)
ACKER-FAIL-STREAM-ID(fail-spout-msgexecutor-data(gettask-datastask-id);;fail
spout-idtuple-finished-infotime-delta)
)))
;;TODO:onfailure,emittupletofailurestream
))))
receive-queue(:receive-queueexecutor-data);;取得receivedisruptorqueue
event-handler(mk-task-receiverexecutor-datatuple-action-fn);;定义disruptor/clojure-handler,使用tuple-action-fn处理从receive-queue里面得到的tuple
has-ackers?(has-ackers?storm-conf)
emitted-count(MutableLong.0)
empty-emit-streak(MutableLong.0)

;;theoverflowbufferisusedtoensurethatspoutsneverblockwhenemitting
;;thisensuresthatthespoutcanalwayscleartheincomingbuffer(acksandfails),which
;;preventsdeadlockfromoccuringacrossthetopology(e.g.Spout->Bolt->Acker->Spout,andall
;;buffersfilledup)
;;whentheoverflowbufferisfull,spoutsstopcallingnextTupleuntilit'sabletocleartheoverflowbuffer
;;thislimitsthesizeoftheoverflowbuffertohowevermanytuplesaspoutemitsinonecallofnextTuple,
;;preventingmemoryissues
overflow-buffer(LinkedList.)]

;;1.3async-loopthread
[(async-loop
(fn[]
;;Iftopologywasstartedininactivestate,don'tcall(.openspout)untilit'sactivatedfirst.
(while(not@(:storm-active-atomexecutor-data))
(Thread/sleep100))

(log-message"Openingspout"component-id":"(keystask-datas))
(doseq[[task-idtask-data]task-datas
:let[^ISpoutspout-obj(:objecttask-data)
tasks-fn(:tasks-fntask-data)

;;1.3.1send-spout-msg
send-spout-msg(fn[out-stream-idvaluesmessage-idout-task-id]
(.incrementemitted-count)
(let[out-tasks(ifout-task-id
(tasks-fnout-task-idout-stream-idvalues);;directgrouping
(tasks-fnout-stream-idvalues));;调用grouper产生targettasks
rooted?(andmessage-idhas-ackers?);;指定messageid并且有acker,说明需要track该message,root?意思需要track的DAG的root
root-id(ifrooted?(MessageId/generateIdrand));;rand.nextLong,随机long,产生root-id
out-ids(fast-list-for[tout-tasks](ifrooted?(MessageId/generateIdrand)))];;对于发送到的每个task,产生一个out-id(out-edgeid)
(fast-list-iter[out-taskout-tasksidout-ids]
(let[tuple-id(ifrooted?
(MessageId/makeRootIdroot-idid);;返回包含hashmap{root-id,out-id}的MessageId对象

(MessageId/makeUnanchored));;返回包含hashmap{}的MessageId对象

out-tuple(TupleImpl.worker-context;;生成tuple对象

values
task-id
out-stream-id
tuple-id)]
(transfer-fnout-task;;调用executor->transfer-fn将tuple发送到spout的发送queue
out-tuple
overflow-buffer)))
(ifrooted?
(do;;如果需要跟踪

(.putpendingroot-id[task-id;;往pendingqueue增加需要track的tuple信息

message-id
{:streamout-stream-id:valuesvalues}
(if(sampler)(System/currentTimeMillis))]);;只有sampler为true,才会设置starttime,后面才会更新metrics和stats
(task/send-unanchoredtask-data;;往ACKER-INIT-STREAM发送message,告诉ackertrack该message
ACKER-INIT-STREAM-ID
[root-id(bit-xor-valsout-ids)task-id]
overflow-buffer))
(whenmessage-id;;rooted?为false,而有message-id,意味着没有acker(has-ackers?为false)
(ack-spout-msgexecutor-datatask-datamessage-id;;既然没有acker,就直接ack
{:streamout-stream-id:valuesvalues}
(if(sampler)0))))
(orout-tasks[]);;send-spout-msg返回值,发送的tasklists或空[]
))]]
(builtin-metrics/register-all(:builtin-metricstask-data)storm-conf(:user-contexttask-data));;注册builtin-metrics

;;1.3.2spout.open
(.openspout-obj
storm-conf
(:user-contexttask-data)
(SpoutOutputCollector.
(reifyISpoutOutputCollector;;实现ISpoutOutputCollector
(^Listemit[this^Stringstream-id^Listtuple^Objectmessage-id];;实现emit
(send-spout-msgstream-idtuplemessage-idnil)
)
(^voidemitDirect[this^intout-task-id^Stringstream-id
^Listtuple^Objectmessage-id]
(send-spout-msgstream-idtuplemessage-idout-task-id)
)
(reportError[thiserror]
(report-errorerror)
)))))
(reset!open-or-prepare-was-called?true)
(log-message"Openedspout"component-id":"(keystask-datas))

;;1.3.3setup-metrics!
(setup-metrics!executor-data);;使用schedule-recurring定期给自己发送METRICS_TICKtuple

(disruptor/consumer-started!(:receive-queueexecutor-data));;设置queue上面的consumerStartedFlag表示consumer已经启动
;;1.3.4fn

(fn[]
;;Thisdesignrequiresthatspoutsbenon-blocking
(disruptor/consume-batchreceive-queueevent-handler);;从recieve-queue取出batchtuples,并使用tuple-action-fn处理

;;trytocleartheoverflow-buffer,将overflow-buffer里面的数据放到发送的缓存queue里面
(try-cause
(while(not(.isEmptyoverflow-buffer))
(let[[out-taskout-tuple](.peekoverflow-buffer)]
(transfer-fnout-taskout-tuplefalsenil)
(.removeFirstoverflow-buffer)))
(catchInsufficientCapacityExceptione
))

(let[active?@(:storm-active-atomexecutor-data)
curr-count(.getemitted-count)]
(if(and(.isEmptyoverflow-buffer);;只有当overflow-buffer为空,并且pending没有达到上限的时候,spout可以继续emittuple
(or(notmax-spout-pending)
(<(.sizepending)max-spout-pending)))
(ifactive?;;storm集群是否active
(do;;stormactive
(when-not@last-active;;如果当前spout出于unactive状态
(reset!last-activetrue)
(log-message"Activatingspout"component-id":"(keystask-datas))
(fast-list-iter[^ISpoutspoutspouts](.activatespout)));;先activespout

(fast-list-iter[^ISpoutspoutspouts](.nextTuplespout)));;调用nextTuple,产生新的tuple
(do;;stormunactive
(when@last-active;;如果spout出于active状态
(reset!last-activefalse)
(log-message"Deactivatingspout"component-id":"(keystask-datas))
(fast-list-iter[^ISpoutspoutspouts](.deactivatespout)));;deactivespout并休眠
;;TODO:logthatit'sgettingthrottled
(Time/sleep100))))
(if(and(=curr-count(.getemitted-count))active?);;没有能够emit新的tuple(前后emitted-count没有变化)
(do(.incrementempty-emit-streak)
(.emptyEmitspout-wait-strategy(.getempty-emit-streak)));;调用spout-wait-strategy进行sleep
(.setempty-emit-streak0)
))
0));;返回0,表示async-loop的sleep时间为0
:kill-fn(:report-error-and-dieexecutor-data)
:factory?true
:thread-namecomponent-id)]))

1.1定义pending

max-spout-pending(executor-max-spout-pendingstorm-conf(counttask-datas))
(defnexecutor-max-spout-pending[storm-confnum-tasks]
(let[p(storm-confTOPOLOGY-MAX-SPOUT-PENDING)]
(ifp(*pnum-tasks))))

/**
*Thestrategyaspoutneedstousewhenitswaiting.Waitingis
*triggeredinoneoftwoconditions:
*
*1.nextTupleemitsnotuples
*2.ThespouthashitmaxSpoutPendingandcan'temitanymoretuples
*
*Thedefaultstrategysleepsforonemillisecond.
*/
publicinterfaceISpoutWaitStrategy{
voidprepare(Mapconf);
voidemptyEmit(longstreak);
}

^ISpoutWaitStrategyspout-wait-strategy(init-spout-wait-strategystorm-conf)

pending(RotatingMap.
2;;microoptimizeforperformanceof.sizemethod,buckets数为2
(reifyRotatingMap$ExpiredCallback
(expire[thismsg-id[task-idspout-idtuple-infostart-time-ms]]
(let[time-delta(ifstart-time-ms(time-delta-msstart-time-ms))]
(fail-spout-msgexecutor-data(gettask-datastask-id)spout-idtuple-infotime-delta)
))))

publicMap<K,V>rotate(){
Map<K,V>dead=_buckets.removeLast();
_buckets.addFirst(newHashMap<K,V>());
if(_callback!=null){
for(Entry<K,V>entry:dead.entrySet()){
_callback.expire(entry.getKey(),entry.getValue());
}
}
returndead;
}