[spark streaming]Receiver数据接收全过程详解

约定不等于承诺〃 2022-05-15 11:04 233阅读 0赞

**一、Receiver启动过程回顾**

![592648-20160526084933256-1760223342.png][]

如图，从ReceiverTracker的start方法开始，调用launchReceivers()方法，给endpoint发送消息，endpoint.send(StartAllReceivers(receivers))，endpoint就是ReceiverTrackerEndpoint，ReceiverTrackerEndpoint调用startReceiver方法在Executor中启动ReceiverSupervisor。详情请阅读链接中的文章[http://blog.csdn.net/zhouzx2010][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，整个数据接收过程如下图：

![592648-20160526084934006-1553524401.png][]

**三、Receiver数据接收过程详解**

3.1 ReceiverSupervisor 启动

Spark Streaming 在集群启动Receiver的时候会先在Executor中启动ReceiverSupervisor来管理Receiver，ReceiverSupervisor的启动是通过调用他的start方法完成的，源码如下：

![592648-20160526084934584-420533165.png][]

ReceiverSupervisor的start方法首先调用了onStart 方法，该方法启动了BlockGenerator，必须在启动Receiver之前调用该方法来启动BlockGenerator，因为必须BlockGenerator先启动，以保证接收到的数据能够被存储起来。该方法在ReceiverSupervisor的子类ReceiverSupervisorImpl有实现：

![592648-20160526084934975-1865202199.png][]

可以看到，在onStart方法中启动了所有的已经注册的BlockGenerator，其中registeredBlockGenerators是ReceiverSupervisorImpl的成员变量，在ReceiverSupervisorImpl实例化是被创建：

![592648-20160526084935319-2127592857.png][]

registeredBlockGenerators中的Generartor是什么时候产生，并加入到registeredBlockGenerators这个队列里的呢？ReceiverSupervisorImpl的createBlockGenerator方法对registeredBlockGenerators进行了元素添加：

![592648-20160526084935834-404310585.png][]

从源码可以看到每一个receiver创建一个BlockGenerator，因为streamId与receiver一一对应。createBlockGenerator首先还将已经停止的BlockGenerator从registeredBlockGenerators队列清除。createBlockGenerator 方法又是在什么时候被调用的呢，答案是在ReceiverSupervisorImpl实例化时候被调用，请看源码：

![592648-20160526084936303-1820672857.png][]

现在registeredBlockGenerators中有了BlockGenerator，接下来分析BlockGenerator的启动过程

3.2 BlockGenerator 启动

![592648-20160526084936850-922985070.png][]

BlockGenerator的start方法启动一个定时器blockIntervalTimer，用来定时生成调用updateCurrentBuffer方法：

![592648-20160526084937334-1597961537.png][]

时间间隔默认是200毫秒：

![592648-20160526084937709-1842450023.png][]

下面看一下updateCurrentBuffer的源码：

![592648-20160526084938116-181885701.png][]

currentBuffer 是一个ArrayBuffer 存储了数据记录，updateCurrentBuffer方法将间隔时间段的数据记录封装成Block，然后将新的Block 放入等待推送的block队列blockForPushing：

![592648-20160526084938709-2067083123.png][]

回到BlockGenerator的start方法的另一个线程blockPushingThread

![592648-20160526084939084-1112846053.png][]

可以看到blockPushingThread的run方法调用了keepPushingBlocks方法将blockForPushing队列中的Block推送给BlockManager：

![592648-20160526084939444-1374265031.png][]

可以看到最多等待10毫秒从blockForPushing中取去Block，然后掉pushBlock方法将block推送给BlockManager，下面看一下pushBlock的源码

![592648-20160526084939959-2083736756.png][]

这里的listener 是 BlockGeneratorListener 对象，在ReceiverSupervisorImpl中被实例化：

![592648-20160526084940366-490523383.png][]

我们关注其中的onPushBlock方法：调用了pushArrayBuffer方法：

![592648-20160526084940741-481972395.png][]

这里调用了一个很重要的方法

pushAndReportBlock

我们看一下源码：

![592648-20160526084941163-1697364615.png][]

pushAndReportBlock方法用来存储接收的的数据Block并将元数据汇报给ReceiverTracker。

Block存储交给receivedBlockHandler处理，receivedBlockHandlery有两种实现方式：

![592648-20160526084941616-1005394609.png][]

看一下BlockManagerBasedBlockHandler的storeBlock方法：

![592648-20160526084942038-758836085.png][]

BlockManagerBasedBlockHandler的storeBlock方法根据ReceivedBlock的不同类型分别处理，最终都是交给了BlockManger管理。

至此BlockGenerator 分析完毕，下面分析 Receiver 的启动过程

**3.3 Receiver 启动**

首先回到ReceiverSupervisorImpl的startReceiver方法：

![592648-20160526084942663-879093049.png][]

首先调用了onReceiverStart方法，将Receiver注册给ReceiverTracker,重点看代码rever.onStart()，调用了receiver的onStart方法，此处我们以SocketReceiver为例进行讲解：

![592648-20160526084943116-786331551.png][]

可以看到SocketReceiver的onStart（）方法先给创建了一个Socket对象赋值给了socket，然后启动了一个后台进行“Socket Receiver” ，在后台进程的run方法中调用了receive（）方法进行数据接收：

![592648-20160526084943709-61910756.png][]

SocketReceiver的receive（）方法不断从socket接收数据，然后调用store方法进行存储。

![592648-20160526084944147-976158647.png][]

此处再次回到ReceiverSupervisor，调用了其中的pushSingle函数：

![592648-20160526084944553-632004655.png][]

最终调用了BlockGenerator的addData方法：

![592648-20160526084945038-637049357.png][]

可以看到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取数据，这样整个数据接收过程不断循环。

![592648-20160526084945741-1436311709.png][]

[592648-20160526084933256-1760223342.png]: /images/20220515/d9b4b1333b084a2fbdd86b04e5a6eb40.png
[http_blog.csdn.net_zhouzx2010]: http://blog.csdn.net/zhouzx2010
[592648-20160526084934006-1553524401.png]: /images/20220515/82ef754b76d64dbe855425aa8cf10446.png
[592648-20160526084934584-420533165.png]: /images/20220515/ea112bf749914139b917fed940415cd1.png
[592648-20160526084934975-1865202199.png]: /images/20220515/bbc678dd54f94112b87828913401c418.png
[592648-20160526084935319-2127592857.png]: /images/20220515/13ee066c8c9748e0b47d7b50dbf712fa.png
[592648-20160526084935834-404310585.png]: /images/20220515/23c17293bc664d3d94fb3d77b16dc962.png
[592648-20160526084936303-1820672857.png]: /images/20220515/47bce83ea2754af1a38e5d51322745a3.png
[592648-20160526084936850-922985070.png]: /images/20220515/91083a2ed0e2485fbb5816fbcb7601a7.png
[592648-20160526084937334-1597961537.png]: /images/20220515/903227c6277d41ed81093713899f695e.png
[592648-20160526084937709-1842450023.png]: /images/20220515/fdf425c398574b098f98beebcef54268.png
[592648-20160526084938116-181885701.png]: /images/20220515/4a388cb2a2274cfe8be42eb5b6dbc91d.png
[592648-20160526084938709-2067083123.png]: /images/20220515/76c4e44e1d024477a098179bd6064858.png
[592648-20160526084939084-1112846053.png]: /images/20220515/4e3daff8f6c84cf88ea32062a34c8b00.png
[592648-20160526084939444-1374265031.png]: /images/20220515/76c83461403f4aa9a1af348d021ddae9.png
[592648-20160526084939959-2083736756.png]: /images/20220515/7bb6651d98724fa092761e544d9f8e83.png
[592648-20160526084940366-490523383.png]: /images/20220515/ec6b086b5636441ab58a9ec516698b66.png
[592648-20160526084940741-481972395.png]: /images/20220515/0ce5d0e62f6a44859892c109847f54af.png
[592648-20160526084941163-1697364615.png]: /images/20220515/7d1eeaac11c74652903a03dbfb9a8736.png
[592648-20160526084941616-1005394609.png]: /images/20220515/853d4b2ec3f9485986eb733b547112c1.png
[592648-20160526084942038-758836085.png]: /images/20220515/fd9eda6fe3f14df8924b9761d78225e8.png
[592648-20160526084942663-879093049.png]: /images/20220515/ba01144275eb4b90b8894f7ab90691ef.png
[592648-20160526084943116-786331551.png]: /images/20220515/c74e4180f7f84fc590edf647913c8736.png
[592648-20160526084943709-61910756.png]: /images/20220515/ed75c618fb0f49d38340e2458bc77092.png
[592648-20160526084944147-976158647.png]: /images/20220515/ff1fd0304f0042b1bd38cc705f1bee62.png
[592648-20160526084944553-632004655.png]: /images/20220515/03e9d15cc55d47bda4c50ebcfa347d73.png
[592648-20160526084945038-637049357.png]: /images/20220515/fa7faf2a53ee4243a4a9c91fa7e96de6.png
[592648-20160526084945741-1436311709.png]: /images/20220515/1b1880d0279d48898112daa1469a3ef9.png