第11课：Spark Streaming源码解读之Driver中的ReceiverTracker架构

本期内容：

1， ReceiverTracker的架构设计

2， 消息循环系统

3， ReceiverTracker具体实现

ReceiverTacker类以下，从源码注释能够看出该类的做用。

    管理ReceiverInputDStreams的执行，记录Receiver发来的元数据信息。ReceiverTacker类构造时必须传入StreamingContext对象。

    ReceiverTacker类内部有ReceiverTackerEndpoint这个消息通讯体，用于和运行在Executor端的ReceiverSupervisorImpl进行通讯，包括Receiver的注册，重启Receiver，清除以前的Block数据，更新限流值，添加Block元数据信息等消息。

    接下来以接收到来自Executor端的ReceiverSupervisorImpl发来添加元数据信息的AddBlock消息，进行讲解具体的处理流程。

        

ReceivedBlockInfo类包含了StreamID，Block中记录条数，元数据Metadata，接收Block的存储结果（BlockID和记录数量）

        

ReceiverBlockTracker类是addBlock方法的具体实现。

        1.调用ReceiverBlockTracker的writeToLog方法

        2.调用ReceiverBlockTracker的getReceivedBlockQueue方法，其中streamIdToUnallocatedBlockQueues为HashMap，Key为StreamID，Value为ReceivedBlockQueue。而ReceivedBlockQueue 的定义为private type ReceivedBlockQueue = mutable.Queue[ReceivedBlockInfo]

        ReceiverBlockTracker类，能够从源码中看出，他会记录全部接收到的Block信息，根据须要把Block分配给Batch。若是设置了checkpoint，开启WAL，则会把全部的操做保存到预写日志中，所以当Driver失败后就能够从checkpoint和WAL中恢复ReceiverTracker的状态。

        ReceiverBlockTracker类中重要的方法，allocateBlocksToBatch。private val timeToAllocatedBlocks = new mutable.HashMap[Time, AllocatedBlocks]存储批处理时刻，分配到的Blocks数据。

        该方法是被ReceiverTracker调用的。

        而ReceiverTracker的allocateBlocksToBatch方法是被JobGenerator的generateJobs方法调用的。

        

ReceiverBlockTracker类中重要的方法，getBlocksOfBatch。

       

 该方法是被ReceiverTracker的getBlocksOfBatch调用。

       

ReceiverTracker的getBlocksOfBatch方法是被ReceiverInputDStream的compute方法调用的。

总结：

Receiver接收到数据，而后合并并存储数据以后，ReceiverSupervisorImpl会把Block的元数据汇报给ReceiverTracker内部的消息通讯体ReceiverTrackerEndpoint。

ReceiverTracker接收到Block的元数据信息以后，由ReceivedBlockTracker管理Block的元数据的分配，JobGenerator会将每一个Batch，从ReceivedBlockTracker中获取属于该Batch的Block元数据信息来生成RDD。

从设计模式来说：ReceiverTrackerEndpoint和ReceivedBlockTracker是门面设计模式，内部实际干事情的是ReceivedBlockTracker，外部通讯体或者表明者就是ReceiverTrackerEndpoint。