Storm源碼分析

第1章　總體架構(gòu)與代碼結(jié)構(gòu)　1
1．1　Storm的總體結(jié)構(gòu)　1
1．2　Storm的元數(shù)據(jù)　3
1．2．1　元數(shù)據(jù)介紹　3
1．2．2　Storm怎么使用這些元數(shù)據(jù)　4
1．3　Storm的代碼結(jié)構(gòu)　7
1．3．1　Clojure代碼　7
1．3．2　Java代碼　8
1．3．3　Trident代碼　9
1．3．4　其他代碼　10
第2章　搭建Storm集群　11
2．1　搭建單機(jī)Storm集群　11
2．2　搭建多機(jī)Storm集群　14
2．2．1　設(shè)置環(huán)境　14
2．2．2　啟動(dòng)Storm集群　15
2．2．3　提交Topology　15
2．3　WordCountTopology介紹　15
2．3．1　RandomSentenceSpout　15
2．3．2　SplitSentence　16
2．3．3　WordCount　17
2．3．4　WordCountTopology構(gòu)建　17
第3章　Storm編程基礎(chǔ)　19
3．1　Fields定義　19
3．2　Tuple接口　20
3．3　常用聲明接口　21
3．3．1　配置聲明接口　22
3．3．2　輸入聲明接口　23
3．3．3　輸出字段聲明接口　24
3．3．4　組件聲明接口　25
3．4　Spout輸出收集器　25
3．4．1　ISpoutOutputCollector和SpoutOutputCollector　25
3．4．2　Executor中ISpoutOutputCollector的實(shí)現(xiàn)　27
3．5　Bolt輸出收集器　28
3．5．1　IOutputCollector和OutputCollector　28
3．5．2　IBasicOutputCollector和BasicOutputCollector　31
3．5．3　BatchOutputCollector和BatchOutputCollectorImpl　32
3．5．4　Executor中的IOutputCollector實(shí)現(xiàn)　34
3．6　組件接口　35
3．7　Spout接口　35
3．7．1　ISpout　36
3．7．2　IRichSpout　38
3．8　Bolt接口　38
3．8．1　IBolt　38
3．8．2　IRichBolt　40
3．8．3　IBasicBolt　40
3．8．4　IBatchBolt　42
3．8．5　小結(jié)　45
3．9　Storm數(shù)據(jù)結(jié)構(gòu)　46
3．9．1　GlobalStreamId　46
3．9．2　消息分組方式　46
3．9．3　StreamInfo　47
3．9．4　ShellComponent　47
3．9．5　ComponentObject　47
3．9．6　ComponentCommon　47
3．9．7　SpoutSpec　48
3．9．8　Bolt　48
3．9．9　StormTopology　49
3．9．10　TopologySummary　49
3．9．11　SupervisorSummary　49
3．9．12　ClusterSummary　50
3．9．13　BoltStats　50
3．9．14　SpoutStats　50
3．9．15　統(tǒng)計(jì)信息　50
3．9．16　DRPC　51
3．10　基本Topology構(gòu)建器　52
3．10．1　TopologyBuilder　52
3．10．2　ConfigGetter　55
3．10．3　SpoutGetter和BoltGetter　55
3．10．4　一個(gè)簡(jiǎn)單例子　56
3．11　異常處理　57
第4章　基礎(chǔ)函數(shù)和工具類　58
4．1　計(jì)時(shí)器　58
4．1．1　mk-timer　58
4．1．2　check-active!　60
4．1．3　schedule　60
4．1．4　schedule-recurring　60
4．1．5　cancel-timer　61
4．2　async-loop　61
4．3　event-manager　62
4．4　even-sampler　63
4．5　ZooKeeper工具類　64
4．5．1　mk-client　64
4．5．2　create-node　65
4．5．3　get-data　65
4．5．4　進(jìn)程內(nèi)啟動(dòng)ZooKeeper　66
4．6　LocalState　66
4．7　ClusterState　68
4．8　StormClusterState　69
第5章　通信機(jī)制　71
5．1　進(jìn)程間通信　71
5．1．1　進(jìn)程間通信協(xié)議　71
5．1．2　LocalCluster模式實(shí)現(xiàn)　72
5．1．3　分布式模式實(shí)現(xiàn)　73
5．1．4　協(xié)議使用　75
5．2　進(jìn)程內(nèi)通信　77
5．2．1　Disruptor Queue的使用　77
5．2．2　DisruptorQueue的Clojure處理器　80
第6章　Nimbus　81
6．1　Nimbus服務(wù)接口定義　81
6．2　Nimbus相關(guān)的數(shù)據(jù)結(jié)構(gòu)　83
6．2．1　Java 數(shù)據(jù)結(jié)構(gòu)　83
6．2．2　Clojure數(shù)據(jù)結(jié)構(gòu)　84
6．3　Nimbus中的線程介紹　86
6．3．1　mk-assignments　87
6．3．2　do-cleanup　89
6．3．3　clean-inbox　90
6．4　Topology狀態(tài)轉(zhuǎn)移　90
6．4．1　transition-name!　90
6．4．2　transition!　91
6．4．3　state-transitions　92
6．5　啟動(dòng)Nimbus服務(wù)　96
6．5．1　launch-server！　96
6．5．2　service-handler　97
6．6　關(guān)閉Nimbus服務(wù)　99
6．7　主要服務(wù)方法　99
6．7．1　submitTopology　99
6．7．2　kill、rebalance、activate、deactivate 方法　101
6．7．3　文件上傳與下載　102
6．7．4　獲取UI所需的信息　104
6．7．3　獲取Topology　106
6．7．6　獲取Storm配置項(xiàng)　107
6．8　主要輔助方法　107
6．8．1　system-topology!　107
6．8．2　normalize-topology　112
6．8．3　compute-new-topology->executor->node+port　114
6．8．4　compute-executors　117
第7章　Scheduler　119
7．1　IScheduler接口　119
7．2　EvenScheduler　120
7．2．1　schedule-topolpgies-evenly　120
7．2．2　schedule-topology　121
7．2．3　get-alive-assigned-node+port->executors　122
7．2．4　sort-slots　123
7．3　DefaultScheduler　124
7．3．1　default-schedule　124
7．3．2　slots-can-reassign　126
7．3．3　bad-slots　126
7．4　IsolationScheduler　127
7．5　調(diào)度示例　131
7．5．1　EvenScheduler和DefaultScheduler　131
7．5．2　IsolationScheduler　134
第8章　Scheduler　137
8．1　與Supervisor相關(guān)的數(shù)據(jù)結(jié)構(gòu)　137
8．1．1　standalone-supervisor　137
8．1．2　Supervisor的數(shù)據(jù)　138
8．1．3　本地存儲(chǔ)數(shù)據(jù)　139
8．2　Supervisor中的線程　140
8．2．1　計(jì)時(shí)器線程　140
8．2．2　同步Nimbus任務(wù)的線程　140
8．2．3　管理Worker進(jìn)程的線程　143
8．3　啟動(dòng)Supervisor　145
8．4　關(guān)閉Supervisor　147
8．5　重要方法介紹　147
8．5．1　launch-worker　147
8．5．2　read-allocated-workers　150
8．5．3　wait-for-worker-launch　151
8．5．4　shutdown-worker　152
8．5．5　download-storm-code　152
第9章　Worker　155
9．1　Worker中的數(shù)據(jù)　155
9．2　Worker中的計(jì)時(shí)器　157
9．2．1　Worker的心跳　157
9．2．2　Executor的心跳　158
9．2．3　Worker中對(duì)ZMQ連接的維護(hù)　159
9．2．4　從ZooKeeper獲取Topology的活躍情況　161
9．2．5　小結(jié)　162
9．3　創(chuàng)建Worker　163
9．4　關(guān)閉Worker　164
9．5　重要輔助方法介紹　165
9．5．1　Worker中的接收函數(shù)　166
9．5．2　Worker中的發(fā)送函數(shù)　167
9．5．3　獲取屬于Worker的Executor　169
9．5．4　創(chuàng)建Executor的接收消息隊(duì)列和查找表　169
9．5．5　下載Topology的配置項(xiàng)以及代碼　170
9．6　小結(jié)　171
第10章　Executor　172
10．1　Executor的數(shù)據(jù)　172
10．2　Executor的輸入和輸出　174
10．2．1　Executor的輸入及處理　174
10．2．2　Executor的輸出及發(fā)送　175
10．3　Spout類型的Executor　176
10．3．1　準(zhǔn)備消息循環(huán)的數(shù)據(jù)　176
10．3．2　Spout輸入處理函數(shù)　178
10．3．3　Spout消息發(fā)送函數(shù)　180
10．3．4　Spout對(duì)象的初始化　181
10．3．5　消息循環(huán)　182
10．4　Bolt類型的Executor　184
10．4．1　準(zhǔn)備消息循環(huán)的數(shù)據(jù)　184
10．4．2　Bolt輸入處理函數(shù)　184
10．4．3　Bolt的消息發(fā)送函數(shù)　185
10．4．4　Bolt對(duì)象的初始化　185
10．4．5　消息循環(huán)　186
10．5　創(chuàng)建Executor　187
10．6　輔助函數(shù)介紹　188
10．6．1　組件的Grouper函數(shù)　188
10．6．2　帶流量控制的錯(cuò)誤報(bào)告方法　193
10．6．3　觸發(fā)系統(tǒng)Ticks　194
10．7　小結(jié)　196
第11章　Task　198
11．1　Task的上下文對(duì)象　198
11．1．1　TopologyContext　198
11．1．2　GeneralTopologyContext　199
11．1．3　WorkerTopologyContext　200
11．1．4　TopologyContext　201
11．2　創(chuàng)建Task數(shù)據(jù)　202
11．3　mk-tasks-fn函數(shù)　204
11．4　send-unanchored　205
11．5　創(chuàng)建Task　206
11．6　Storm中傳輸?shù)南⒁约靶蛄谢?06
第12章　Storm的Ack框架　208
12．1　Acker Bolt的實(shí)現(xiàn)分析　209
12．2　啟動(dòng)消息跟蹤　211
12．3　消息跟蹤　212
12．4　Ack機(jī)制的例子　214
第13章　系統(tǒng)運(yùn)行統(tǒng)計(jì)　216
13．1　基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)以及更新算法　216
13．1．1　滑動(dòng)窗口的數(shù)據(jù)結(jié)構(gòu)　216
13．1．2　滑動(dòng)窗口的回調(diào)函數(shù)　220
13．1．3　滑動(dòng)窗口集合的類型　221
13．2　Storm中的統(tǒng)計(jì)信息　222
13．2．1　Stats中定義的統(tǒng)計(jì)類別　222
13．2．2　運(yùn)行統(tǒng)計(jì)的更新　223
13．2．3　運(yùn)行統(tǒng)計(jì)的更新時(shí)間點(diǎn)　223
13．2．4　獲取統(tǒng)計(jì)數(shù)據(jù)　228
13．3　運(yùn)行統(tǒng)計(jì)的Thrift結(jié)構(gòu)　229
第14章　系統(tǒng)運(yùn)行統(tǒng)計(jì)的另一種實(shí)現(xiàn)　231
14．1　內(nèi)置統(tǒng)計(jì)信息的計(jì)算　231
14．1．1　MultiCountMetric　232
14．1．2　MultiReducedMetric　233
14．2　內(nèi)置統(tǒng)計(jì)類型　234
14．2．1　Spout類型的內(nèi)置統(tǒng)計(jì)　235
14．2．2　Bolt類型的內(nèi)置統(tǒng)計(jì)　235
14．3　統(tǒng)計(jì)觸發(fā)消息　235
14．3．1　注冊(cè)統(tǒng)計(jì)信息　236
14．3．2　觸發(fā)消息的產(chǎn)生與發(fā)送　237
14．3．3　處理統(tǒng)計(jì)觸發(fā)消息　238
14．4　運(yùn)行統(tǒng)計(jì)收集節(jié)點(diǎn)　239
14．5　SystemBolt　241
第15章　事務(wù)Topology的實(shí)現(xiàn)　243
15．1　事務(wù)Topology的實(shí)現(xiàn)概述　243
15．1．1　事務(wù)Topology的類型　244
15．1．2　事務(wù)Topology的類關(guān)系　245
15．2　ITransactionalSpout接口　246
15．3　協(xié)調(diào)Spout節(jié)點(diǎn)的執(zhí)行器　248
15．3．1　ZooKeeper客戶端工具　248
15．3．2　協(xié)調(diào)Spout的執(zhí)行器　255
15．3．3　消息發(fā)送Bolt的執(zhí)行器　261
15．4　CoordinatedBolt的實(shí)現(xiàn)分析　264
15．4．1　TrackingInfo　264
15．4．2　CoordinatedOutput-Collector　265
15．4．3　CoordinatedBolt中的消息類型　267
15．4．4　成員變量以及主要方法分析　267
15．5　分區(qū)的事務(wù)類型　271
15．5．1　分區(qū)的事務(wù)Spout接口　271
15．5．2　分區(qū)的事務(wù)Spout的執(zhí)行器　273
15．6　分區(qū)的模糊事務(wù)Spout　277
15．6．1　分區(qū)的模糊事務(wù)Spout的接口　277
15．6．2　模糊的事務(wù)Spout執(zhí)行器　278
15．7　事務(wù)Topology的構(gòu)建器　281
15．7．1　構(gòu)建器的構(gòu)造函數(shù)及成員變量　281
15．7．2　設(shè)置Bolt對(duì)象　283
15．7．3　構(gòu)建Topology　284
15．7．4　輸入流聲明器　286
第16章　事務(wù)Topology示例　288
16．1　例子代碼　288
16．1．1　分區(qū)的事務(wù)Spout　288
16．1．2　局部計(jì)數(shù)Bolt的實(shí)現(xiàn)　291
16．1．3　全局計(jì)數(shù)Bolt的實(shí)現(xiàn)　292
16．2　構(gòu)建Topology　293
16．3　事務(wù)處理示例　295
第17章　Trident的Spout節(jié)點(diǎn)　298
17．1　ITridentSpout接口　298
17．1．1　BatchCoordinator接口　299
17．1．2　TridentSpoutCoordinator　300
17．1．3　MasterBatchCoordinator　301
17．1．4　消息發(fā)送節(jié)點(diǎn)接口　306
17．1．5　消息發(fā)送接口的執(zhí)行器　306
17．2　適配IRichSpout接口　307
17．3　適配IBatchSpout接口　311
17．4　Trident中分區(qū)的Spout類型　311
17．4．1　分區(qū)Spout接口　311
17．4．2　分區(qū)Spout的執(zhí)行器　313
17．5　模糊事務(wù)類型的Spout節(jié)點(diǎn)　316
17．5．1　模糊事務(wù)類型的Spout接口　317
17．5．2　模糊事務(wù)類型Spout的執(zhí)行器　317
17．6　構(gòu)建Spout節(jié)點(diǎn)　320
17．6．1　TridentTopology的newStream調(diào)用　320
17．6．2　TridentTopology中newDRPCStream調(diào)用　321
第18章　Trident的存儲(chǔ)　322
18．1　存儲(chǔ)的基本接口　322
18．2　MapState接口的實(shí)現(xiàn)　323
18．2．1　非事務(wù)類型的存儲(chǔ)　324
18．2．2　事務(wù)類型的存儲(chǔ)　325
18．2．3　模糊事務(wù)類型存儲(chǔ)　327
18．3　值的序列化方法　329
18．4　數(shù)據(jù)更新接口　330
18．4．1　CombinerValueUpdater　330
18．4．2　ReducerValueUpdater　331
18．5　存儲(chǔ)更新接口　331
18．5．1　ReducerAggStateUpdater　332
18．5．2　MapReducerAggStateUpdater　332
18．5．3　BaseStateUpdater　334
18．6　創(chuàng)建存儲(chǔ)對(duì)象　334
第19章　Trident消息　336
19．1　ValuePointer　336
19．2　Factory接口及其實(shí)現(xiàn)　337
19．2．1　ProjectionFactory　338
19．2．2　FreshOutputFactory　339
19．2．3　OperationOutputFactory　339
19．2．4　RootFactory　341
19．3　消息工廠的例子　342
19．4　TridentTupleView　342
19．5　ComboList　343
第20章　Trident操作與處理節(jié)點(diǎn)　346
20．1　操作的基本接口　346
20．2　Aggregator實(shí)現(xiàn)　347
20．2．1　GroupedAggregator　348
20．2．2　ChainedAggregatorImpl　350
20．2．3　SingleEmitAggregator　353
20．3　用戶接口及其實(shí)現(xiàn)　355
20．3．1　ReducerAggregator接口及其實(shí)現(xiàn)　355
20．3．2　CombinerAggregator接口及其實(shí)現(xiàn)　356
20．4　所有處理節(jié)點(diǎn)的上下文　357
20．4．1　單個(gè)處理節(jié)點(diǎn)的上下文　358
20．4．2　操作執(zhí)行的上下文　359
20．5　Trident的輸出收集器　359
20．5．1　FreshCollector　359
20．5．2　CaptureCollector　360
20．5．3　GroupCollector　360
20．5．4　AppendCollector　361
20．5．5　AddIdCollector　361
20．6　Trident的處理節(jié)點(diǎn)　362
20．6．1　TridentProcessor接口　363
20．6．2　PartitionPersistProcessor　363
20．6．3　StateQueryProcessor　365
20．7　聚集器的執(zhí)行　367
第21章　Trident流的基本操作　370
21．1　流的成員變量和基礎(chǔ)方法　370
21．1．1　流的成員變量　370
21．1．2　流節(jié)點(diǎn)名字　370
21．1．3　流的映射檢查　372
21．1．4　添加節(jié)點(diǎn)　372
21．2　流映射操作　373
21．3　流的分組操作　374
21．4　流的逐行操作　374
21．5　流的分區(qū)操作　374
21．6　流的單聚集器聚集操作　376
21．7　流的多聚集器聚集操作　377
21．7．1　ChainedAggregatorDeclarer　377
21．7．2　分區(qū)上的局部聚集操作　379
21．7．3　全局聚集操作　379
21．7．4　含有多個(gè)聚集器的partitionAggregate操作　381
21．8　流的聚集操作　382
21．9　流的分區(qū)寫入操作　383
21．10　查詢操作　384
21．11　流的全局寫入操作　384
21．12　流的操作與有向圖構(gòu)建　384
21．13　分組流　385
21．13．1　成員變量　385
21．13．2　逐行操作　385
21．13．3　分組流的分區(qū)聚集操作　386
21．13．4　查詢操作　386
21．13．5　聚集操作　386
21．13．6　寫入操作　387
21．14　利用流操作來構(gòu)建Topology的例子　388
第22章　Trident中流的交互操作　392
22．1　基本接口　392
22．2　JoinerMultiReducer　393
22．2．1　成員變量及構(gòu)造函數(shù)　393
22．2．2　execute方法　395
22．2．3　complete方法　397
22．3　GroupedMultiReducerExecutor　397
22．4　MultiReducerProcessor　399
22．5　連接操作　401
22．6　流合并操作　403
第23章　Trident中的Bolt節(jié)點(diǎn)　404
23．1　SubTopologyBolt　404
23．1．1　輸入準(zhǔn)備　404
23．1．2　成員變量　405
23．1．3　主要方法　406
23．2　Trident中的Bolt執(zhí)行器　409
23．2．1　ITridentBatchBolt接口　410
23．2．2　TrackedBatch　410
23．2．3　定制的輸出收集器　412
23．2．4　消息類型　414
23．2．5　數(shù)據(jù)成員分析　414
23．2．6　主要成員方法分析　416
第24章　Trident的執(zhí)行優(yōu)化　420
24．1　節(jié)點(diǎn)類型　420
24．1．1　基本節(jié)點(diǎn)類型　420
24．1．2　Spout節(jié)點(diǎn)　422
24．1．3　處理節(jié)點(diǎn)　422
24．1．4　分區(qū)節(jié)點(diǎn)　423
24．2　執(zhí)行優(yōu)化算法　426
24．2．1　節(jié)點(diǎn)組　426
24．2．2　節(jié)點(diǎn)組的合并算法　427
24．2．3　處理節(jié)點(diǎn)組中的分區(qū)節(jié)點(diǎn)　431
24．2．4　節(jié)點(diǎn)組以不同的方式收聽相同流　431
24．2．5　執(zhí)行優(yōu)化后的節(jié)點(diǎn)組　434
24．2．6　計(jì)算節(jié)點(diǎn)組的并行度　434
第25章　Trident與DRPC　437
25．1　DRPC服務(wù)器　438
25．1．1　DRPC服務(wù)器的成員變量　438
52．1．2　DRPC用戶接口及其實(shí)現(xiàn)　439
25．1．3　DRPC Topology端接口及其實(shí)現(xiàn)　440
25．1．4　啟動(dòng)DRPC服務(wù)器　441
25．2　DRPC的客戶端　442
25．3　DRPC中Spout節(jié)點(diǎn)　443
25．4　DRPC Spout的執(zhí)行器　446
25．5　completeDRPC操作　449
25．6　返回DRPC結(jié)果　451
第26章　Trident的Topology構(gòu)建器　453
26．1　基本工具函數(shù)　453
26．1．1　committerBatches　453
26．1．2　fleshOutStreamBatchIds　453
26．1．3　getOutputStreamBatchGroups　454
26．2　TridentTopologyBuilder　455
26．2．1　成員變量　455
26．2．2　設(shè)置Spout節(jié)點(diǎn)　456
26．2．3　設(shè)置Bolt節(jié)點(diǎn)　458
26．3　一個(gè)例子　460
第27章　多語言　462
27．1　ShellProcess　462
27．2　ShellBolt　464
27．2．1　成員變量　464
27．2．2　讀寫線程　465
27．3　ShellSpout　467
第28章　Storm中的配置項(xiàng)　469