SparkStreaming作业提交源码分析接收数据篇我们通过KafkaUtils.createStream函数可以创建KafkaReceiver类（这是默认的KafkaReceiver，如果spark.streaming.receiver.writeAheadLog.enable配置选项设置为true，则会使用ReliableKafkaReceiver，其中会使用WAL机制来保证数据的可靠性，也就是保证数据不丢失。）在KafkaReceiver类中首先会在onStart方法中初始化一些环境，比如创建Consumer(这个就是用来从Kafka的Topic中读取消息的消费者)。在初始化完相关环境之后会在线程池中启动MessageHandler来从Kafka中接收数据：//HandlesKafkamessagesprivateclassMessageHandler(stream:KafkaStream[K,V])extendsRunnable{defrun(){logInfo("StartingMessageHandler.")try{valstreamIterator=stream.iterator()while(streamIterator.hasNext()){valmsgAndMetadata=streamIterator.next()store((msgAndMetadata.key,msgAndMetadata.message))}}catch{casee:Throwable=>logError("Errorhandlingmessage;exiting",e)}}}该线程负责从Kafka中读取数据，并将读取到的数据存储到BlockGenerator中（通过调用store方法实现），msgAndMetadata.key其实就是Topic的Key值；而msgAndMetadata.message就是我们要的消息。store函数是Receiver类提供的，所有继承自该类的子类(KafkaReceiver、ActorReceiver、ReliableKafkaReceiver等)都拥有该方法。其内部的实现是调用了blockGenerator的addData方法，最终是将数据存储在currentBuffer中，而currentBuffer其实就是一个ArrayBuffer[Any]。在blockGenerator内部存在两个线程：（1）、定期地生成新的batch，然后再将之前生成的batch封装成block。这里的定期其实就是spark.streaming.blockInterval参数配置的。（2）、将生成的block发送到BlockManager中。第一个线程定期地调用updateCurrentBuffer函数将存储在currentBuffer中的数据封装成Block，然后放在blocksForPushing中，blocksForPushing是ArrayBlockingQueue[Block]类型的队列，其大小默认是10，我们可以通过spark.streaming.blockQueueSize参数配置（当然，在很多情况下这个值不需要我们去配置）。当blocksForPushing没有多余的空间，那么该线程就会阻塞，直到有剩余的空间可用于存储新生成的Block。如果你的数据量真的很大，大到blocksForPushing无法及时存储那些block，这时候你就得考虑加大spark.streaming.blockQueueSize的大小了。updateCurrentBuffer函数的实现如下：/**Changethebuffertowhichsinglerecordsareaddedto.*/privatedefupdateCurrentBuffer(time:Long):Unit=synchronized{try{valnewBlockBuffer=currentBuffercurrentBuffer=newArrayBuffer[Any]if(newBlockBuffer.size>0){valblockId=StreamBlockId(receiverId,time-blockInterval)valnewBlock=newBlock(blockId,newBlockBuffer)listener.onGenerateBlock(blockId)blocksForPushing.put(newBlock)//putisblockingwhenqueueisfulllogDebug("Lastelementin"+block