在上一篇中介紹了Receiver的整體架構和設計原理，本篇內容主要介紹Receiver在Executor中數據接收和存儲過程
一、Receiver啟動過程回顧
如圖，
從ReceiverTracker的start方法開始，調用launchReceivers()方法，給endpoint發送消息，endpoint.send(StartAllReceivers(receivers))，endpoint就是ReceiverTrackerEndpoint，
ReceiverTrackerEndpoint調用startReceiver方法在Executor中啟動ReceiverSupervisor。詳情請閱讀鏈接中的文章
http://blog.csdn.net/zhouzx2010
二、Receiver數據接收全過程總覽
Receiver數據接收核心成員：
1.ReceiverSupervisor
2.BlockGenerator
3.Receiver
4.ReceiverTracker
5.BlockManager
首先
ReceiverSupervisor啟動，繼而啟動BlockGenerator和Receiver，其中
BlockGenerator首先啟動一個定時器定時將接收的數據封裝成Block，然后啟動一個線程不斷將Block數據推送給BlockManager。而
Receiver啟動后不斷接收數據，並不斷將接受的數據添加到BlockGenerator中，這樣
BlockGenerator就不斷將
Receiver接收的數據推送到BlockManager
，整個數據接收過程如下圖：
三、Receiver數據接收過程詳解
3.1 ReceiverSupervisor 啟動
Spark Streaming 在集群啟動Receiver的時候會先在Executor中啟動ReceiverSupervisor來管理Receiver，
ReceiverSupervisor的啟動是通過調用他的start方法完成的，源碼如下：
ReceiverSupervisor的start方法首先調用了onStart 方法，該方法啟動了
BlockGenerator，
必須
在啟動Receiver之前調用該方法來啟動BlockGenerator，因為必須BlockGenerator先啟動，以保證接收到的數據能夠被存儲起來。該
方法在
ReceiverSupervisor的子類
ReceiverSupervisorImpl
有實現
：
可以看到，在onStart方法中啟動了所有的已經注冊的BlockGenerator，其中registeredBlockGenerators是
ReceiverSupervisorImpl的成員變量，在
ReceiverSupervisorImpl實例化是被創建：
registeredBlockGenerators中的Generartor是什么時候產生，並加入到
registeredBlockGenerators這個隊列里的呢？
ReceiverSupervisorImpl
的
createBlockGenerator方法對
registeredBlockGenerators進行了元素添加
：
從源碼可以看到
每一個receiver創建一個BlockGenerator，因為streamId與
receiver
一一對應。
createBlockGenerator首先還將已經停止的BlockGenerator從
registeredBlockGenerators隊列清除。
createBlockGenerator 方法又是在什么時候被調用的呢，答案是在
ReceiverSupervisorImpl實例化時候被調用，請看源碼：
現在
registeredBlockGenerators中有了
BlockGenerator，接下來分析
BlockGenerator的啟動過程
3.2 BlockGenerator 啟動
BlockGenerator的start方法
啟動一個定時器blockIntervalTimer，用來定時生成調用
updateCurrentBuffer方法：
時間間隔默認是200毫秒：
下面看一下
updateCurrentBuffer的源碼：
currentBuffer 是一個ArrayBuffer 存儲了數據記錄，updateCurrentBuffer方法將間隔時間段的數據記錄封裝成Block，然后將新的Block 放入等待推送的block隊列blockForPushing：
回到BlockGenerator的start方法的另一個線程blockPushingThread
可以看到
blockPushingThread的run方法調用了keepPushingBlocks方法將
blockForPushing隊列中的Block推送給BlockManager：
可以看到最多等待10毫秒從
blockForPushing中取去Block，然后掉pushBlock方法將block推送給BlockManager，下面看一下
pushBlock的源碼
這里的listener 是 BlockGeneratorListener 對象，在ReceiverSupervisorImpl中被實例化：
我們關注其中的onPushBlock方法：調用了pushArrayBuffer方法：
這里調用了一個很重要的方法
pushAndReportBlock
我們看一下源碼：
pushAndReportBlock方法用來存儲接收的的數據Block並將元數據匯報給ReceiverTracker。
Block存儲交給receivedBlockHandler處理，receivedBlockHandlery有兩種實現方式：
看一下BlockManagerBasedBlockHandler的storeBlock方法：
BlockManagerBasedBlockHandler的storeBlock方法根據ReceivedBlock的不同類型分別處理，最終都是交給了BlockManger管理。
至此BlockGenerator 分析完畢，下面分析 Receiver 的啟動過程
3.3 Receiver 啟動
首先回到
ReceiverSupervisorImpl的startReceiver方法：
首先調用了onReceiverStart方法，將Receiver注冊給ReceiverTracker,重點看代碼rever.onStart()，調用了receiver的onStart方法，此處我們以SocketReceiver為例進行講解：
可以看到
SocketReceiver的onStart（）方法先給創建了一個Socket對象賦值給了socket，然后啟動了一個后台進行“Socket Receiver” ，在后台進程的run方法中調用了receive（）方法進行數據接收：
SocketReceiver的receive（）方法不斷從socket接收數據，然后調用store方法進行存儲。
此處再次回到ReceiverSupervisor，調用了其中的pushSingle函數：
最終調用了BlockGenerator的addData方法：
可以看到
BlockGenerator的addData方法將數據添加到currentBuffer中，然后blockIntervalTimer會每個200毫秒從
currentBuffer取出數據封裝成Block，並將Block添加到blockForPushing隊列，blockPushingThread會將
blockForPushing隊列中的block推送給BlockManager
。至此Receiver運行流程分析完畢。
四、總結：receiver數據接收流程圖
總體過程：ReceiverSupervisor啟動時，啟動了BlockGenerator和Receiver，其中
BlockGenerator首先啟動一個定時器定時將接收的數據封裝成Block，然后啟動一個線程不斷將Block數據推送給BlockManager。而
Receiver啟動后不斷接收數據，不斷將接受的數據添加到BlockGenerator的currentBuffer中，
BlockGenerator不斷從
currentBuffer取數據，這樣整個數據接收過程不斷循環。
