Spark 订制版：011~Spark Streaming源码解读之Driver中的ReceiverTracker架构设计以及具体实现彻底研究
本讲内容：
a. ReceiverTracker的架构设计
b. 消息循环系统
c. ReceiverTracker具体实现
注：本讲内容基于Spark 1.6.1版本（在2016年5月来说是Spark最新版本）讲解。
上节回顾
上一讲中，我们主要给大家介绍Spark Streaming在接收数据的全生命周期贯通；
a. 当有Spark Streaming有应用程序的时候Spark Streaming会持续不断的接收数据
b. 一般Receiver和Driver不在一个进程中的，所以接收到数据之后要不断的汇报给Driver
c. Spark Streaming要接收数据肯定要使用消息循环器，循环器不断的接收到数据之后，然后将数据存储起来，再将存储完的数据汇报给Driver
d. Sparkstreaming接收数据的整个流程类似于MVC模式，M就是Receiver，V就是Driver，C就是ReceiverSupervisor
e. ReceiverSupervisor是控制器，Receiver的启动是靠ReceiverTracker启动的，Receiver接收到数据之后是靠ReceiverSupervisor存储数据的。然后Driver就获得元数据也就是界面，通过界面来操作底层的数据，这个元数据就相当于指针
开讲
上节课讲到了Receiver是如何不断的接收数据的，并且接收到的数据的元数据会汇报给ReceiverTracker
本讲我们看看ReceiverTracker具体的架构及其功能、源码实现
ReceiverTracker的架构设计
a. ReceiverTracker以Driver中具体的算法在具体的Executor之上启动Receiver，而且启动Receiver的方式是把每个Receiver封装成一个Task， 此时一个Job中就一个Task，而Task中就一条数据，也就是Receiver数据，实质上说，ReceiverTracker启动Receiver之时就会封装在一个个Job，有多个Job就有多个Receiver，即有多个Receiver启动就有多个Job封装
b. ReceiverTracker在启动Receiver的时候，有一个Receiversupervisor其里面有一个ReceiversupervisorImpl实现类， Receiversupervisor实际上启动之时就启动了Receiver，Receiver不断的接收数据，通过BlockGenerator把自已“接收的数据”变成一个个的Block。然后在时间定时器的作用下会不断的把数据存储（此时存储有2种方式，第一种是通过BlockManager方式存储，另一种先写日志Write，通过WAL方式），数据存储之后ReceiverSupervisorImpl会把存储后的数据的元数据Metadate汇报给ReceiverTracker,其实是汇报给ReceiverTracker中的RPC实体ReceiverTrackerEndpoint
c. ReceiverTracker用来管理Receiver中的数据执行，数据执行层面包括Receiver的启动、回收、执行过程中接收数据的管理，当然也包括“Receiver”的容错
当ReceiverSupervisorImpl将元数据汇报给ReceiverTracker，ReceiverTracker接收数据之后，是怎么对数据进行管理呢？
就让我们走进源码一探究竟吧
首先，我们来看看ReceiverSupervisorImpl类
receivedBlockHandler方法，是用来写数据
PushAndReportBlock存储Block数据，且把信息汇报给Driver
上面代码中，trackerEndpoint其实就是ReceiverTracker中的ReceiverTrackerEndpoint
PushAndReportBlock中， ReceivedBlockInfo:封装Block的存储信息
上面代码中，ReceivedBlockStoreResult，存储Blocks
ReceiverTracker的源码遍历
Receiver和ReceiverTracker之间通信的是怎么完成的？
而Driver和ReceiverTrackerEndpoint之间的交流是通过ReceiverTrackerLocalMessage进行的
我们来进入到ReceiverTrackerLocalMessage的子类中
ReceiverTracker：管理Receiver的启动，Receiver的执行，回收，执行过程中接收数据的管理。DStreamGraph中会有成员记录所有的数据流来源，免得数据会丢失
ReceiverTracker中的receiveAndReply方法
进入到receiveAndReply方法中
进入上面代码中addBlock源码
进入上面代码中addBlock源码
由上面的代码可知道：把具体的一个Receiver汇报上来的数据的元数据信息写入streamIdToUnallocatedBlockQueues中
还可以看出getReceivedBlockQueue是ReceivedBlockQueue类型
进入到streamIdToUnallocatedBlockQueues源码中
上面代码中HashMap中第一个参数是StreamId，第二个参数ReceivedBlockQueue是StreamId对应接收到的Receiver
继续ReceiverBlockTracker源码分析
保持跟踪所有接收到的Block。并且根据需要把他们分配给batches.
假设提供checkpoint的话，ReceiverBlockTracker中的信息包括receiver接收到的block数据和分配的信息。Driver如果失败的话，就读取checkpoint中的信息。
ReceivedBlockTracker通过调用allocateBlocksToBatch方法把接收到的数据分配给当前执行的Batch Duractions作业（allocateBlocksToBatch被JobGenerator调用的）
进入到上面AllocatedBlocks方法源码中
（JobGenerator中的generateJobs）
再次回到ReceiverTracker中的receive方法
在ReceiverSupervisorImpl的receive方法中就接收到了ReceiverTracker的CleanupOldBlocks消息
我们进入到RateLimiter中的updateRate方法
ReceiverTracker中receiveAndReply中StopAllReceivers源码解析
进入到stopReceivers源码中
在ReceiverSupervisorImpl中receive接收到了上面代码中 receiverTrackingInfos.values.flatMap(_.endpoint).foreach { _.send(StopReceiver) }发送的消息
上面代码中的stop函数是在ReceiverSupervisor类中实现的
继续进入到上面代码中stopReceiver源码
最终调用了上面代码中的receiver.onStop()方法
ReceiverSupervisor类中的onReceiverStop方法在子类ReceiverSupervisorImpl中会有具体实现
ReceiverSupervisor类中的receiver.onStop()方法在ReceiverSupervisorImpl中实现如下
由此我们得知：
a. Receiver接收到数据之后合并存储数据后，ReceiverSupervisorImpl会把数据汇报给ReceiverTracker, ReceiverTracker接收到元数据，其内部汇报的是RPC通信体，接收到数据之后，内部有ReceivedBlockTracker会管理数据的分配，JobGenerator会将每个Batch,每次工作的时候会根据元数据信息从ReceiverTracker中获取相应的元数据信息生成RDD。
b.ReceiverBlockTracker中 allocateBlocksToBatch专门管理Block元数据信息，作为一个内部的管理对象。
门面设计模式
从设计模式来讲：ReceiverTrackerEndpoint和ReceivedBlockTracker是门面设计模式，。
ReceiverTracker和ReceivedBlockTracker的关系是：内部实际干事情的是ReceivedBlockTracker，外部通信体或者代表者就是ReceiverTracker
JobGenerator在计算基于Batch的Job的时候，以及DStreamGraph生成RDD的DAG的时候会调用此方法
当一个Batch计算完的时候，他会把已经使用的数据块的数据信息清理掉
备注：
1、DT大数据梦工厂微信公众号DT_Spark
2、Spark专家：王家林
3、新浪微博: http://www.weibo.com/ilovepains