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

1. 交由 BlockManager 根據(jù)設(shè)置的 StorageLevel 存入 executor 的內(nèi)存或磁盤中
2. 通過 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í)行以下操作：

1. 將 currentBuffer 賦值給 newBlockBuffer
2. 將 currentBuffer 指向新的空的 ArrayBuffer 對象
3. 將 newBlockBuffer 封裝成 newBlock
4. 將 newBlock 添加到 blocksForPushing 隊列中

blockIntervalMs 由 spark.streaming.blockInterval 控制，默認(rèn)是 200ms。

blockPushingThread & blocksForPushing & blockQueueSize

blocksForPushing 是一個定長數(shù)組，長度由 blockQueueSize 決定，默認(rèn)為10，可通過 spark.streaming.blockQueueSize 改變。上面分析到，blockIntervalTimer中的線程會定時將 block 塞入該隊列。

還有另一條線程不斷送該隊列中取出 block，然后調(diào)用 ReceiverSupervisorImpl.pushArrayBuffer(...) 來將 block 存儲，這條線程就是blockPushingThread。

PS: blocksForPushing為ArrayBlockingQueue類型。ArrayBlockingQueue是一個阻塞隊列，能夠自定義隊列大小，當(dāng)插入時，如果隊列已經(jīng)沒有空閑位置，那么新的插入線程將阻塞到該隊列，一旦該隊列有空閑位置，那么阻塞的線程將執(zhí)行插入

以上，通過分析各個成員，也說明了 BlockGenerator 是如何存儲單條數(shù)據(jù)的。