ReceiverSupervisorImpl共提供了4個將從 receiver 傳遞過來的數(shù)據(jù)轉(zhuǎn)換成 block 并存儲的方法拔创，分別是：
- pushSingle: 處理單條數(shù)據(jù)
- pushArrayBuffer: 處理數(shù)組形式數(shù)據(jù)
- pushIterator: 處理 iterator 形式處理
- pushBytes: 處理 ByteBuffer 形式數(shù)據(jù)
其中崔涂，pushArrayBuffer涤姊、pushIterator屯掖、pushBytes最終調(diào)用pushAndReportBlock裸删；而pushSingle將調(diào)用defaultBlockGenerator.addData(data)噪径，我們分別就這兩種形式做說明
pushAndReportBlock
我們針對存儲 block 簡化 pushAndReportBlock 后的代碼如下：
def pushAndReportBlock(
receivedBlock: ReceivedBlock,
metadataOption: Option[Any],
blockIdOption: Option[StreamBlockId]
) {
...
val blockId = blockIdOption.getOrElse(nextBlockId)
receivedBlockHandler.storeBlock(blockId, receivedBlock)
...
}
首先獲取一個新的 blockId棍掐，之后調(diào)用 receivedBlockHandler.storeBlock, receivedBlockHandler 在 ReceiverSupervisorImpl 構(gòu)造函數(shù)中初始化藏雏。當(dāng)啟用了 checkpoint 且 spark.streaming.receiver.writeAheadLog.enable 為 true 時，receivedBlockHandler 被初始化為 WriteAheadLogBasedBlockHandler 類型作煌；否則將初始化為 BlockManagerBasedBlockHandler類型掘殴。
WriteAheadLogBasedBlockHandler#storeBlock 將 ArrayBuffer, iterator, bytes 類型的數(shù)據(jù)序列化后得到的 serializedBlock
- 交由 BlockManager 根據(jù)設(shè)置的 StorageLevel 存入 executor 的內(nèi)存或磁盤中
- 通過 WAL 再存儲一份
而BlockManagerBasedBlockHandler#storeBlock將 ArrayBuffer, iterator, bytes 類型的數(shù)據(jù)交由 BlockManager 根據(jù)設(shè)置的 StorageLevel 存入 executor 的內(nèi)存或磁盤中，并不再通過 WAL 存儲一份
pushSingle
pushSingle將調(diào)用 BlockGenerator#addData(data: Any) 通過積攢的方式來存儲數(shù)據(jù)粟誓。接下來對 BlockGenerator 是如何積攢一條一條數(shù)據(jù)最后寫入 block 的邏輯奏寨。
上圖為 BlockGenerator 的各個成員，首選對各個成員做介紹：
currentBuffer
變長數(shù)組鹰服，當(dāng) receiver 接收的一條一條的數(shù)據(jù)將會添加到該變長數(shù)組的尾部
- 可能會有一個 receiver 的多個線程同時進(jìn)行添加數(shù)據(jù)服爷，這里是同步操作
- 添加前，會由 rateLimiter 檢查一下速率获诈，是否加入的速度過快仍源。如果過快的話就需要 block 住，等到下一秒再開始添加舔涎。最高頻率由
spark.streaming.receiver.maxRate控制笼踩，默認(rèn)值為Long.MaxValue，具體含義是單個 Receiver 每秒鐘允許添加的條數(shù)亡嫌。
blockIntervalTimer & blockIntervalMs
分別是定時器和時間間隔嚎于。blockIntervalTimer中有一個線程，每隔blockIntervalMs會執(zhí)行以下操作：
- 將 currentBuffer 賦值給 newBlockBuffer
- 將 currentBuffer 指向新的空的 ArrayBuffer 對象
- 將 newBlockBuffer 封裝成 newBlock
- 將 newBlock 添加到 blocksForPushing 隊(duì)列中
blockIntervalMs 由 spark.streaming.blockInterval 控制挟冠，默認(rèn)是 200ms于购。
blockPushingThread & blocksForPushing & blockQueueSize
blocksForPushing 是一個定長數(shù)組，長度由 blockQueueSize 決定知染，默認(rèn)為10肋僧，可通過 spark.streaming.blockQueueSize 改變。上面分析到，blockIntervalTimer中的線程會定時將 block 塞入該隊(duì)列嫌吠。
還有另一條線程不斷送該隊(duì)列中取出 block止潘，然后調(diào)用 ReceiverSupervisorImpl.pushArrayBuffer(...) 來將 block 存儲，這條線程就是blockPushingThread辫诅。
PS: blocksForPushing為ArrayBlockingQueue類型凭戴。ArrayBlockingQueue是一個阻塞隊(duì)列，能夠自定義隊(duì)列大小炕矮，當(dāng)插入時么夫，如果隊(duì)列已經(jīng)沒有空閑位置，那么新的插入線程將阻塞到該隊(duì)列肤视，一旦該隊(duì)列有空閑位置档痪，那么阻塞的線程將執(zhí)行插入
以上，通過分析各個成員钢颂，也說明了 BlockGenerator 是如何存儲單條數(shù)據(jù)的钞它。
