class NettyBlockTransferService

这个类的主要目的是 和其他 executor 节点 交互block data。交互block data 包括上传和下载2个方面。所以在这个类里面会有一个 server 的服务，因为要把其他节点的block data 存放到 本节点的 BlockManager中，所以这个类与BlockManager有交互，可以通过阅读源码看出，在BlockManager中会持有一个 NettyBlockTransferService的实例，NettyBlockTransferService的初始化方法中也需要传入BlockManager的实例 。
下面来看看 NettyBlockTransferService 的源码：

// NettyBlockTransferService 这个类的主要作用是 使用 Netty 获取和上传 blocks
private[spark] class NettyBlockTransferService(conf: SparkConf,securityManager: SecurityManager,bindAddress: String,override val hostName: String, //本机的 hostname_port: Int, // 默认是 0 由系统分配 端口numCores: Int) //指定的核数extends BlockTransferService {// TODO: Don't use Java serialization, use a more cross-version compatible serialization format.private val serializer = new JavaSerializer(conf)private val authEnabled = securityManager.isAuthenticationEnabled() //一般测试环境为 falseprivate val transportConf: TransportConf = SparkTransportConf.fromSparkConf(conf, "shuffle", numCores)private[this] var transportContext: TransportContext = _private[this] var server: TransportServer = _private[this] var clientFactory: TransportClientFactory = _private[this] var appId: String = _//初始化方法 会在 BlockManager 的initialize 方法中 调用 本类的init 方法override def init(blockDataManager: BlockDataManager): Unit = {val rpcHandler = new NettyBlockRpcServer(conf.getAppId, serializer, blockDataManager) //这个是service 处理请求的具体handlervar serverBootstrap: Option[TransportServerBootstrap] = Nonevar clientBootstrap: Option[TransportClientBootstrap] = Noneif (authEnabled) { //falseserverBootstrap = Some(new AuthServerBootstrap(transportConf, securityManager))clientBootstrap = Some(new AuthClientBootstrap(transportConf, conf.getAppId, securityManager))}transportContext = new TransportContext(transportConf, rpcHandler)clientFactory = transportContext.createClientFactory(clientBootstrap.toSeq.asJava)server = createServer(serverBootstrap.toList) //启动 serverappId = conf.getAppId //application IDlogInfo(s"Server created on ${hostName}:${server.getPort}") // server.getPort 就是 系统自动分配的 port}/** Creates and binds the TransportServer, possibly trying multiple ports. */private def createServer(bootstraps: List[TransportServerBootstrap]): TransportServer = {def startService(port: Int): (TransportServer, Int) = {val server = transportContext.createServer(bindAddress, port, bootstraps.asJava)(server, server.getPort)}Utils.startServiceOnPort(_port, startService, conf, getClass.getName)._1}override def shuffleMetrics(): MetricSet = {require(server != null && clientFactory != null, "NettyBlockTransferServer is not initialized")new MetricSet {val allMetrics = new JHashMap[String, Metric]()override def getMetrics: JMap[String, Metric] = {allMetrics.putAll(clientFactory.getAllMetrics.getMetrics)allMetrics.putAll(server.getAllMetrics.getMetrics)allMetrics}}}// get Blocks 从那台机器上获取哪些block ，在 BlockManager 的 getRemoteBytes 方法中会使用到这个方法override def fetchBlocks(host: String,port: Int,execId: String,blockIds: Array[String],listener: BlockFetchingListener,tempFileManager: DownloadFileManager): Unit = {logTrace(s"Fetch blocks from $host:$port (executor id $execId)")try {val blockFetchStarter = new RetryingBlockFetcher.BlockFetchStarter {override def createAndStart(blockIds: Array[String], listener: BlockFetchingListener) {val client = clientFactory.createClient(host, port)new OneForOneBlockFetcher(client, appId, execId, blockIds, listener,transportConf, tempFileManager).start() //注意 这里，会到 OneForOneBlockFetcher 类中，}}val maxRetries = transportConf.maxIORetries()if (maxRetries > 0) {// Note this Fetcher will correctly handle maxRetries == 0; we avoid it just in case there's// a bug in this code. We should remove the if statement once we're sure of the stability.new RetryingBlockFetcher(transportConf, blockFetchStarter, blockIds, listener).start()} else {blockFetchStarter.createAndStart(blockIds, listener)}} catch {case e: Exception =>logError("Exception while beginning fetchBlocks", e)blockIds.foreach(listener.onBlockFetchFailure(_, e))}}override def port: Int = server.getPort//上传 哪些 blockId 到哪台机器，在BlockManager 的 replicate 方法中 会使用到override def uploadBlock(hostname: String,port: Int,execId: String,blockId: BlockId,blockData: ManagedBuffer,level: StorageLevel,classTag: ClassTag[_]): Future[Unit] = {val result = Promise[Unit]()val client = clientFactory.createClient(hostname, port)// StorageLevel and ClassTag are serialized as bytes using our JavaSerializer.// Everything else is encoded using our binary protocol.val metadata = JavaUtils.bufferToArray(serializer.newInstance().serialize((level, classTag))) //metadata 主要是 StorageLevel和 classTag// Convert or copy nio buffer into array in order to serialize it.val array = JavaUtils.bufferToArray(blockData.nioByteBuffer())//这里是 上传一个 block 到 其他的节点，对应的处理 服务端的处理 在 NettyBlockRpcServer 的 receive 的 case uploadBlock 中client.sendRpc(new UploadBlock(appId, execId, blockId.name, metadata, array).toByteBuffer,new RpcResponseCallback {override def onSuccess(response: ByteBuffer): Unit = {logTrace(s"Successfully uploaded block $blockId")result.success((): Unit)}override def onFailure(e: Throwable): Unit = {logError(s"Error while uploading block $blockId", e)result.failure(e)}})result.future}override def close(): Unit = {if (server != null) {server.close()}if (clientFactory != null) {clientFactory.close()}}
}

class NettyBlockRpcServer

这个类是NettyBlockTransferService的service处理 请求的handler 里面会有 上传和下载block data的具体处理方法，下面来看看源码：

//这个类是 NettyBlockTransferService RPC 响应处理类
class NettyBlockRpcServer(appId: String,serializer: Serializer,blockManager: BlockDataManager)extends RpcHandler with Logging {private val streamManager = new OneForOneStreamManager()//这个方法是 真正的处理 rpc请求的方法override def receive(client: TransportClient, //主要用来 发送 请求rpcMessage: ByteBuffer, //接收到的消息responseContext: RpcResponseCallback): Unit = { // rpc 需要 response 的上下文val message: BlockTransferMessage = BlockTransferMessage.Decoder.fromByteBuffer(rpcMessage)logTrace(s"Received request: $message")message match {case openBlocks: OpenBlocks => //这里处理 别的节点  get Blocks 的请求val blocksNum = openBlocks.blockIds.lengthval blocks: SeqView[ManagedBuffer, Seq[_]] = for (i <- (0 until blocksNum).view)yield blockManager.getBlockData(BlockId.apply(openBlocks.blockIds(i)))  //使用 BlockManager 中的方法 获取 这个节点的 block dataval streamId = streamManager.registerStream(appId, blocks.iterator.asJava,client.getChannel)logTrace(s"Registered streamId $streamId with $blocksNum buffers")responseContext.onSuccess(new StreamHandle(streamId, blocksNum).toByteBuffer)case uploadBlock: UploadBlock => //处理 别的节点的 block上传 请求// StorageLevel and ClassTag are serialized as bytes using our JavaSerializer.val (level: StorageLevel, classTag: ClassTag[_]) = { // 首先 拿到 StorageLevel 和 classTag 元信息serializer.newInstance().deserialize(ByteBuffer.wrap(uploadBlock.metadata)).asInstanceOf[(StorageLevel, ClassTag[_])]}val data = new NioManagedBuffer(ByteBuffer.wrap(uploadBlock.blockData)) //再拿到 block dataval blockId = BlockId(uploadBlock.blockId)blockManager.putBlockData(blockId, data, level, classTag) //最后 根据 StorageLevel再使用  本节点的 blockManager put 到内存 或者 磁盘中去responseContext.onSuccess(ByteBuffer.allocate(0)) //返回 处理成功标志}}override def getStreamManager(): StreamManager = streamManager
}

OneForOneBlockFetcher

这个类的作用是 包装了 fetche 其他节点的 block data的 rpc调用方法。

//点对点 Block 获取block器
public class OneForOneBlockFetcher {private static final Logger logger = LoggerFactory.getLogger(OneForOneBlockFetcher.class);private final TransportClient client;private final OpenBlocks openMessage;private final String[] blockIds; //获取哪些 block 为什么 这个数据类型是 String 啦？ org.apache.spark.storage.BlockId 的伴随对象中 apply 方法就是 将String类型的block 转化为真正的 Blcok 子类private final BlockFetchingListener listener;private final ChunkReceivedCallback chunkCallback;private final TransportConf transportConf;private final DownloadFileManager downloadFileManager;private StreamHandle streamHandle = null;public OneForOneBlockFetcher(TransportClient client,String appId,String execId,String[] blockIds,BlockFetchingListener listener,TransportConf transportConf) {this(client, appId, execId, blockIds, listener, transportConf, null);}public OneForOneBlockFetcher(TransportClient client, // request 客户端String appId,String execId,String[] blockIds,BlockFetchingListener listener,TransportConf transportConf,DownloadFileManager downloadFileManager) {this.client = client;this.openMessage = new OpenBlocks(appId, execId, blockIds); //组装this.blockIds = blockIds;this.listener = listener;this.chunkCallback = new ChunkCallback();this.transportConf = transportConf;this.downloadFileManager = downloadFileManager;}/** Callback invoked on receipt of each chunk. We equate a single chunk to a single block. */private class ChunkCallback implements ChunkReceivedCallback {@Overridepublic void onSuccess(int chunkIndex, ManagedBuffer buffer) {// On receipt of a chunk, pass it upwards as a block.listener.onBlockFetchSuccess(blockIds[chunkIndex], buffer);}@Overridepublic void onFailure(int chunkIndex, Throwable e) {// On receipt of a failure, fail every block from chunkIndex onwards.String[] remainingBlockIds = Arrays.copyOfRange(blockIds, chunkIndex, blockIds.length);failRemainingBlocks(remainingBlockIds, e);}}/*** Begins the fetching process, calling the listener with every block fetched.* The given message will be serialized with the Java serializer, and the RPC must return a* {@link StreamHandle}. We will send all fetch requests immediately, without throttling.*/public void start() {if (blockIds.length == 0) {throw new IllegalArgumentException("Zero-sized blockIds array");}//openMessage 就是 new OpenBlocks(appId, execId, blockIds);client.sendRpc(openMessage.toByteBuffer(), new RpcResponseCallback() { //那么这个 rpc消息发送之后，将会由 org.apache.spark.network.netty.NettyBlockRpcServer 的 receive 处理 的 case openBlocks/*** 下面的是 处理返回的 信息* @param response*/@Overridepublic void onSuccess(ByteBuffer response) {try {streamHandle = (StreamHandle) BlockTransferMessage.Decoder.fromByteBuffer(response);logger.trace("Successfully opened blocks {}, preparing to fetch chunks.", streamHandle);// Immediately request all chunks -- we expect that the total size of the request is// reasonable due to higher level chunking in [[ShuffleBlockFetcherIterator]].for (int i = 0; i < streamHandle.numChunks; i++) {if (downloadFileManager != null) { // downloadFileManager 一般为 null，只有当 block size 超过一定长度之后 才会 不为 nullclient.stream(OneForOneStreamManager.genStreamChunkId(streamHandle.streamId, i),new DownloadCallback(i));} else {client.fetchChunk(streamHandle.streamId, i, chunkCallback);}}} catch (Exception e) {logger.error("Failed while starting block fetches after success", e);failRemainingBlocks(blockIds, e);}}@Overridepublic void onFailure(Throwable e) {logger.error("Failed while starting block fetches", e);failRemainingBlocks(blockIds, e);}});}/** Invokes the "onBlockFetchFailure" callback for every listed block id. */private void failRemainingBlocks(String[] failedBlockIds, Throwable e) {for (String blockId : failedBlockIds) {try {listener.onBlockFetchFailure(blockId, e);} catch (Exception e2) {logger.error("Error in block fetch failure callback", e2);}}}private class DownloadCallback implements StreamCallback {private DownloadFileWritableChannel channel = null;private DownloadFile targetFile = null;private int chunkIndex;DownloadCallback(int chunkIndex) throws IOException {this.targetFile = downloadFileManager.createTempFile(transportConf);this.channel = targetFile.openForWriting();this.chunkIndex = chunkIndex;}@Overridepublic void onData(String streamId, ByteBuffer buf) throws IOException {while (buf.hasRemaining()) {channel.write(buf);}}@Overridepublic void onComplete(String streamId) throws IOException {listener.onBlockFetchSuccess(blockIds[chunkIndex], channel.closeAndRead());if (!downloadFileManager.registerTempFileToClean(targetFile)) {targetFile.delete();}}@Overridepublic void onFailure(String streamId, Throwable cause) throws IOException {channel.close();// On receipt of a failure, fail every block from chunkIndex onwards.String[] remainingBlockIds = Arrays.copyOfRange(blockIds, chunkIndex, blockIds.length);failRemainingBlocks(remainingBlockIds, cause);targetFile.delete();}}
}