原創(chuàng)文章，轉(zhuǎn)載請注明作者信息和文章鏈接
在上一篇中介紹了Receiver的整體架構(gòu)和設(shè)計原理贷腕，本篇內(nèi)容主要介紹Receiver在Executor中數(shù)據(jù)接收和存儲過程
一拧烦、Receiver啟動過程回顧
如圖，從ReceiverTracker的start方法開始贞间，調(diào)用launchReceivers()方法挠唆，給endpoint發(fā)送消息处窥，endpoint.send(StartAllReceivers(receivers))，endpoint就是ReceiverTrackerEndpoint玄组，ReceiverTrackerEndpoint調(diào)用startReceiver方法在Executor中啟動ReceiverSupervisor滔驾。詳情請閱讀鏈接中的文章http://blog.csdn.net/zhouzx2010
二、Receiver數(shù)據(jù)接收全過程總覽
Receiver數(shù)據(jù)接收核心成員：
1.ReceiverSupervisor
2.BlockGenerator
3.Receiver
4.ReceiverTracker
5.BlockManager
首先ReceiverSupervisor啟動俄讹，繼而啟動BlockGenerator和Receiver哆致，其中BlockGenerator首先啟動一個定時器定時將接收的數(shù)據(jù)封裝成Block，然后啟動一個線程不斷將Block數(shù)據(jù)推送給BlockManager患膛。而Receiver啟動后不斷接收數(shù)據(jù)摊阀，并不斷將接受的數(shù)據(jù)添加到BlockGenerator中，這樣BlockGenerator就不斷將Receiver接收的數(shù)據(jù)推送到BlockManager，整個數(shù)據(jù)接收過程如下圖：
三胞此、Receiver數(shù)據(jù)接收過程詳解
3.1 ReceiverSupervisor 啟動
Spark Streaming 在集群啟動Receiver的時候會先在Executor中啟動ReceiverSupervisor來管理Receiver臣咖，ReceiverSupervisor的啟動是通過調(diào)用他的start方法完成的，源碼如下：
ReceiverSupervisor的start方法首先調(diào)用了onStart 方法豌鹤，該方法啟動了BlockGenerator亡哄，必須在啟動Receiver之前調(diào)用該方法來啟動BlockGenerator枝缔，因為必須BlockGenerator先啟動布疙，以保證接收到的數(shù)據(jù)能夠被存儲起來。該方法在ReceiverSupervisor的子類ReceiverSupervisorImpl有實現(xiàn)：
可以看到愿卸，在onStart方法中啟動了所有的已經(jīng)注冊的BlockGenerator灵临，其中registeredBlockGenerators是ReceiverSupervisorImpl的成員變量，在ReceiverSupervisorImpl實例化是被創(chuàng)建：
registeredBlockGenerators中的Generartor是什么時候產(chǎn)生趴荸，并加入到registeredBlockGenerators這個隊列里的呢儒溉？ReceiverSupervisorImpl的createBlockGenerator方法對registeredBlockGenerators進行了元素添加：
從源碼可以看到每一個receiver創(chuàng)建一個BlockGenerator，因為streamId與receiver一一對應(yīng)发钝。createBlockGenerator首先還將已經(jīng)停止的BlockGenerator從registeredBlockGenerators隊列清除顿涣。createBlockGenerator 方法又是在什么時候被調(diào)用的呢，答案是在ReceiverSupervisorImpl實例化時候被調(diào)用酝豪，請看源碼：
現(xiàn)在registeredBlockGenerators中有了BlockGenerator涛碑，接下來分析BlockGenerator的啟動過程
3.2 BlockGenerator 啟動
BlockGenerator的start方法啟動一個定時器blockIntervalTimer，用來定時生成調(diào)用updateCurrentBuffer方法：
時間間隔默認是200毫秒：
下面看一下updateCurrentBuffer的源碼：
currentBuffer 是一個ArrayBuffer 存儲了數(shù)據(jù)記錄孵淘，updateCurrentBuffer方法將間隔時間段的數(shù)據(jù)記錄封裝成Block蒲障，然后將新的Block 放入等待推送的block隊列blockForPushing：
回到BlockGenerator的start方法的另一個線程blockPushingThread
可以看到blockPushingThread的run方法調(diào)用了keepPushingBlocks方法將blockForPushing隊列中的Block推送給BlockManager：
可以看到最多等待10毫秒從blockForPushing中取去Block，然后掉pushBlock方法將block推送給BlockManager瘫证，下面看一下pushBlock的源碼
這里的listener 是 BlockGeneratorListener 對象揉阎，在ReceiverSupervisorImpl中被實例化：
我們關(guān)注其中的onPushBlock方法：調(diào)用了pushArrayBuffer方法：
這里調(diào)用了一個很重要的方法
pushAndReportBlock
我們看一下源碼：
pushAndReportBlock方法用來存儲接收的的數(shù)據(jù)Block并將元數(shù)據(jù)匯報給ReceiverTracker。
Block存儲交給receivedBlockHandler處理背捌，receivedBlockHandlery有兩種實現(xiàn)方式：
看一下BlockManagerBasedBlockHandler的storeBlock方法：
BlockManagerBasedBlockHandler的storeBlock方法根據(jù)ReceivedBlock的不同類型分別處理毙籽，最終都是交給了BlockManger管理。
至此BlockGenerator 分析完畢毡庆，下面分析 Receiver 的啟動過程
3.3 Receiver 啟動
首先回到ReceiverSupervisorImpl的startReceiver方法：
首先調(diào)用了onReceiverStart方法惧财，將Receiver注冊給ReceiverTracker,重點看代碼rever.onStart()，調(diào)用了receiver的onStart方法扭仁，此處我們以SocketReceiver為例進行講解：
可以看到SocketReceiver的onStart（）方法先給創(chuàng)建了一個Socket對象賦值給了socket垮衷，然后啟動了一個后臺進行“Socket Receiver” ，在后臺進程的run方法中調(diào)用了receive（）方法進行數(shù)據(jù)接收：
SocketReceiver的receive（）方法不斷從socket接收數(shù)據(jù)乖坠，然后調(diào)用store方法進行存儲搀突。
此處再次回到ReceiverSupervisor，調(diào)用了其中的pushSingle函數(shù)：
最終調(diào)用了BlockGenerator的addData方法：
可以看到BlockGenerator的addData方法將數(shù)據(jù)添加到currentBuffer中熊泵，然后blockIntervalTimer會每個200毫秒從currentBuffer取出數(shù)據(jù)封裝成Block仰迁，并將Block添加到blockForPushing隊列甸昏，blockPushingThread會將blockForPushing隊列中的block推送給BlockManager。至此Receiver運行流程分析完畢徐许。
四施蜜、總結(jié)：receiver數(shù)據(jù)接收流程圖
總體過程：ReceiverSupervisor啟動時，啟動了BlockGenerator和Receiver雌隅，其中BlockGenerator首先啟動一個定時器定時將接收的數(shù)據(jù)封裝成Block翻默，然后啟動一個線程不斷將Block數(shù)據(jù)推送給BlockManager。而Receiver啟動后不斷接收數(shù)據(jù)恰起，不斷將接受的數(shù)據(jù)添加到BlockGenerator的currentBuffer中修械，BlockGenerator不斷從currentBuffer取數(shù)據(jù)，這樣整個數(shù)據(jù)接收過程不斷循環(huán)检盼。
原創(chuàng)文章肯污，轉(zhuǎn)載請注明作者信息和文章鏈接
