原创文章，转载请注明: 转载自 周岳飞的博客( http://blog.csdn.net/zhouzx2010)
在上一篇中介绍了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取数据，这样整个数据接收过程不断循环。
原创文章，转载请注明: 转载自 听风居士博客( http://blog.csdn.net/zhouzx2010)