private[netty] val transportConf = SparkTransportConf.fromSparkConf(...)private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)private val streamManager = new NettyStreamManager(this)private val transportContext = new TransportContext(transportConf,　　new NettyRpcHandler(dispatcher, this, streamManager))//用于创建TransportClient的工厂类private val clientFactory = transportContext.createClientFactory(createClientBootstraps())//volatile 关键字保证变量在多线程之间的可见性@volatile private var server: TransportServer = _

package org.apache.spark.rpcimport org.apache.spark.SparkException//创建RpcEnv的工厂类，必须有一个空构造函数才能通过反射创建private[spark] trait RpcEnvFactory {  def create(config: RpcEnvConfig): RpcEnv}private[spark] trait RpcEndpoint {   //当前RpcEndpoint所属的RpcEnv   val rpcEnv: RpcEnv     //获取RpcEndpoint相关联的RpcEndpointRef  final def self: RpcEndpointRef = {    require(rpcEnv != null, "rpcEnv has not been initialized")    rpcEnv.endpointRef(this)  }  //接收消息并处理，不回复客户端  def receive: PartialFunction[Any, Unit] = {    case _ => throw new SparkException(self + " does not implement 'receive'")  }  //接收消息并处理，通过RpcCallContext回复客户端  def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {    case _ => context.sendFailure(new SparkException(self + " won't reply anything"))  }  //onError、onConnected、onDisconnected、onNetworkError、onStart、onStop顾名思义  //用于停止当前RpcEndpoint，注意onStop是trait定义的抽象方法，在停止RpcEndpoint时调用，做一些收尾工作  final def stop(): Unit = {    val _self = self    if (_self != null) {      rpcEnv.stop(_self)    }  }}//线程安全的、串行处理消息的ThreadSafeRpcEndpointprivate[spark] trait ThreadSafeRpcEndpoint extends RpcEndpoint

private[spark] abstract class RpcEndpointRef(conf: SparkConf)  extends Serializable with Logging {  //rpc最大重连次数，默认3，可使用spark.rpc.numRetries属性配置  private[this] val maxRetries = RpcUtils.numRetries(conf)  //rpc每次重连等待的毫秒数，默认3s，可使用spark.rpc.retry.wait属性配置  private[this] val retryWaitMs = RpcUtils.retryWaitMs(conf)  //rpc的ask操作默认超时时间，默认120s，可使用spark.rpc.askTimeout(优先级高)/spark.network.timeout属性配置  private[this] val defaultAskTimeout = RpcUtils.askRpcTimeout(conf)    //返回当前RpcEndpointRef对应的RpcEndpoint的RPC地址  def address: RpcAddress  //返回当前RpcEndpointRef对应的RpcEndpoint的名称  def name: String  //发送单向异步的消息到相应的RpcEndpoint.receive。  def send(message: Any): Unit  //发送一条消息到相应的RpcEndpoint.receiveAndReply，并在指定的超时内接收处理结果。此方法只发送消息一次，从不重试。  def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]  def ask[T: ClassTag](message: Any): Future[T] = ask(message, defaultAskTimeout)  //发送同步请求到相应的RpcEndpoint.receiveAndReply，并在超时时间内等待处理结果，当抛出异常时会请求重试次数以内的重连。  def askSync[T: ClassTag](message: Any): T = askSync(message, defaultAskTimeout)  def askSync[T: ClassTag](message: Any, timeout: RpcTimeout): T = {    val future = ask[T](message, timeout)    timeout.awaitResult(future)  }}

//通过SparkTransportConf的fromSparkConf方法来构建TransportConf需要三个参数：sparkConf、模块名module和可用内核数private[netty] val transportConf = SparkTransportConf.fromSparkConf(    //先克隆SparkConf并设置节点间取数据的连接数    conf.clone.set("spark.rpc.io.numConnectionsPerPeer", "1"),    //设置模块名    "rpc",    //Netty传输线程数，如果小于或等于0，线程数就是系统可用处理器的数量，最多为8线程。    conf.getInt("spark.rpc.io.threads", numUsableCores))

private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)

def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {    //使用当前RpcEndpoint所在的NettyRpcEnv的地址和RpcEndpoint的名称创建RpcEndpointAddress对象    val addr = RpcEndpointAddress(nettyEnv.address, name)    //创建RpcEndpoint的引用对象    val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)    synchronized {      if (stopped) {        throw new IllegalStateException("RpcEnv has been stopped")      }      if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {        throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")      }      //创建EndpointData并放入endpoints缓存      val data = endpoints.get(name)      //将RpcEndpoint与NettyRpcEndpointRef映射关系放入endpointRefs缓存      endpointRefs.put(data.endpoint, data.ref)      //将EndpointData放入阻塞队列receivers,由于EndpointData是新建的，内部会新建Inbox并执行Inbox的主构造函数，      //向Inbox自身的messages列表中放入OnStart消息，MessageLoop线程会取出此EndpointData并调用当前Inbox的process方法      //处理OnStart消息，启动与此Inbox相关联的Endpoint。      receivers.offer(data)  // for the OnStart message    }    endpointRef  }

private val threadpool: ThreadPoolExecutor = {    //获取可用处理器数，numUsableCores是NettyRpcEnv的入参，如果大于0则等于numUsableCores，否则为当前系统可用处理器    val availableCores =      if (numUsableCores > 0) numUsableCores else Runtime.getRuntime.availableProcessors()    //获取当前线程池的大小，默认为2和可用处理器之间的最大值，可用spark.rpc.netty.dispatcher.numThreads属性配置    val numThreads = nettyEnv.conf.getInt("spark.rpc.netty.dispatcher.numThreads",      math.max(2, availableCores))    //创建线程池    val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")    //启动多个线程运行MessageLoop任务    for (i <- 0 until numThreads) {      pool.execute(new MessageLoop)    }    //返回线程池引用    pool  }  /** Message loop used for dispatching messages. */  private class MessageLoop extends Runnable {    override def run(): Unit = {      try {        while (true) {          try {            //在阻塞队列中获取EndpointData            val data = receivers.take()            //如果EndpointData是空数据，则将它重新放回队列并直接返回，这样可以让其他MessageLoop获取到这个空EndpointData并结束线程 　　　　　　　　//private val PoisonPill = new EndpointData(null,null,null)             if (data == PoisonPill) {              // Put PoisonPill back so that other MessageLoops can see it.              receivers.offer(PoisonPill)              return            }            //调用inbox的process方法对消息进行处理            data.inbox.process(Dispatcher.this)          } catch {            case NonFatal(e) => logError(e.getMessage, e)          }        }      } catch {        case ie: InterruptedException => // exit      }    }  }

def process(dispatcher: Dispatcher): Unit = {    var message: InboxMessage = null    //线程并发检查，如果不允许多线程执行且当前激活线程不为0，直接返回    inbox.synchronized {      if (!enableConcurrent && numActiveThreads != 0) {        return      }      //获取消息，如果消息不为空，则当前激活线程+1，否则return返回      message = messages.poll()      if (message != null) {        numActiveThreads += 1      } else {        return      }    }        while (true) {      //对匹配执行时可能发生的错误，使用Endpoint的onError方法处理      safelyCall(endpoint) {        //匹配不同类型的消息进行处理        message match{...}    }        //对激活进程数量的控制，如果不允许多线程处理且当前激活进程不为1，当前线程退出，numActiveThreads - 1      //如果message为空，没有消息需要处理，当前线程退出，numActiveThreads - 1      inbox.synchronized {        // "enableConcurrent" will be set to false after `onStop` is called, so we should check it        // every time.        if (!enableConcurrent && numActiveThreads != 1) {          // If we are not the only one worker, exit          numActiveThreads -= 1          return        }        message = messages.poll()        if (message == null) {          numActiveThreads -= 1          return        }      }    }  }

def stop(rpcEndpointRef: RpcEndpointRef): Unit = {    synchronized {      if (stopped) {        // This endpoint will be stopped by Dispatcher.stop() method.        return      }      unregisterRpcEndpoint(rpcEndpointRef.name)    }  }

private def unregisterRpcEndpoint(name: String): Unit = {    //取出EndpointData    val data = endpoints.remove(name)    if (data != null) {      //调用Inbox的stop方法      data.inbox.stop()      //将EndpointData重新放入receivers队列，让MessageLoop线程能读取到Stop状态，进行相应的处理      receivers.offer(data)  // for the OnStop message    }    // Don't clean `endpointRefs` here because it's possible that some messages are being processed    // now and they can use `getRpcEndpointRef`. So `endpointRefs` will be cleaned in Inbox via    // `removeRpcEndpointRef`.  }

/*   *  当要移除一个EndpointData时，其Inbox可能正在对消息进行处理，所以调用Inbox的stop方法平滑过渡处理;   *  将允许并发运行设置为false，并设置当前Inbox为stopped状态，将当前Inbox所属的EndpointData重新放入receivers，   *  Inbox.process方法会匹配执行相应的处理，调用Dispatcher.removeRpcEndpointRef方法从endpointRefs缓存中移除当前RpcEndpointRef的映射； 　　*  在匹配执行OnStop消息的最后，会调用RpcEndpoint的OnStop方法停止RpcEndpoint。   */  def stop(): Unit = inbox.synchronized {    // The following codes should be in `synchronized` so that we can make sure "OnStop" is the last    // message    if (!stopped) {        // We should disable concurrent here. Then when RpcEndpoint.onStop is called, it's the only        // thread that is processing messages. So `RpcEndpoint.onStop` can release its resources        // safely.      enableConcurrent = false      stopped = true      messages.add(OnStop)      // Note: The concurrent events in messages will be processed one by one.    }  }

def stop(): Unit = {    synchronized {      if (stopped) {        return      }      stopped = true    }    // Stop all endpoints. This will queue all endpoints for processing by the message loops.    //调用unregisterRpcEndpoint方法，对Dispatcher中的所有EndpointData进行去注册，会向endpoints中每个EndpointData中的Inbox中放置    //OnStop消息；最后向receivers中投放PoisonPill，即空EndpointData，以停止所有的MessageLoop线程    endpoints.keySet().asScala.foreach(unregisterRpcEndpoint)    // Enqueue a message that tells the message loops to stop.    receivers.offer(PoisonPill)    threadpool.shutdown()  }

/**    *   将消息提交给指定的RpcEndpoint    * @param endpointName         endpoint名称    * @param message              消息类型    * @param callbackIfStopped  endpoint为stop状态时的回调函数    */  private def postMessage(      endpointName: String,      message: InboxMessage,      callbackIfStopped: (Exception) => Unit): Unit = {    val error = synchronized {      //从endpoints缓存获取EndpointData      val data = endpoints.get(endpointName)      if (stopped) {        Some(new RpcEnvStoppedException())      } else if (data == null) {        Some(new SparkException(s"Could not find $endpointName."))      } else {        //如果endpointData不是停止状态且endpoints缓存中确实有这个EndpointData        //调用对应的Inbox.post将消息加入Inbox的消息列表中        data.inbox.post(message)        //将EndpointData加入receivers队列，以便MessageLoop线程处理此Inbox中的消息        receivers.offer(data)        None      }    }    // We don't need to call `onStop` in the `synchronized` block    error.foreach(callbackIfStopped)  }

//在Inbox未停止时，将message加入messages缓存def post(message: InboxMessage): Unit = inbox.synchronized {  if (stopped) {    // We already put "OnStop" into "messages", so we should drop further messages    onDrop(message)  } else {    messages.add(message)    false  }}

private val transportContext = new TransportContext(transportConf,    new NettyRpcHandler(dispatcher, this, streamManager))

private[netty] class NettyRpcHandler(    dispatcher: Dispatcher,    nettyEnv: NettyRpcEnv,    streamManager: StreamManager) extends RpcHandler with Logging {  // A variable to track the remote RpcEnv addresses of all clients  private val remoteAddresses = new ConcurrentHashMap[RpcAddress, RpcAddress]()  //带回调函数的receive方法，调用internalReceive方法将将ByteBuffer类型的消息转化为RequestMessage  //最后调用dispatcher.postRemoteMessage将消息投递到Inbox，由RpcEndpoint处理消息并回复客户端  override def receive(      client: TransportClient,      message: ByteBuffer,      callback: RpcResponseCallback): Unit = {    val messageToDispatch = internalReceive(client, message)    dispatcher.postRemoteMessage(messageToDispatch, callback)  }  //方法重载，RpcEndpoint处理完消息不会回复客户端  override def receive(      client: TransportClient,      message: ByteBuffer): Unit = {    val messageToDispatch = internalReceive(client, message)    dispatcher.postOneWayMessage(messageToDispatch)  }  //将ByteBuffer类型的消息转化为RequestMessage  private def internalReceive(client: TransportClient, message: ByteBuffer): RequestMessage = {    //从TransportClient中获取远端地址RpcAddress    val addr = client.getChannel().remoteAddress().asInstanceOf[InetSocketAddress]    assert(addr != null)    val clientAddr = RpcAddress(addr.getHostString, addr.getPort)    //封装消息    val requestMessage = RequestMessage(nettyEnv, client, message)    //如果没有发送者地址信息，使用从TransportClient获取的远端地址RpcAddress、消息的接收者(RpcEndpoint)、消息内容构造新的消息    if (requestMessage.senderAddress == null) {      // Create a new message with the socket address of the client as the sender.      new RequestMessage(clientAddr, requestMessage.receiver, requestMessage.content)    } else {      // The remote RpcEnv listens to some port, we should also fire a RemoteProcessConnected for      // the listening address      //获取发送者地址信息，将远端地址RpcAddress和发送者地址信息映射关系放入缓存remoteAddresses      val remoteEnvAddress = requestMessage.senderAddress      if (remoteAddresses.putIfAbsent(clientAddr, remoteEnvAddress) == null) {        //向endpoints缓存中的所有EndpointData的Inbox中放入RemoteProcessConnected类型的消息        dispatcher.postToAll(RemoteProcessConnected(remoteEnvAddress))      }      requestMessage    }  }     ... //其他类型消息的处理，与receive类似 }

//用于处理请求超时的调度器val timeoutScheduler = ThreadUtils.newDaemonSingleThreadScheduledExecutor("netty-rpc-env-timeout")//用于异步处理客户端创建的线程池private[netty] val clientConnectionExecutor = ThreadUtils.newDaemonCachedThreadPool(    "netty-rpc-connection",    conf.getInt("spark.rpc.connect.threads", 64))    /**  *    缓存远端RPC地址与OutBox的关系  *    OutBox与之前的Inbox类似，Outbox是在客户端使用，通过OutboxMessage封装对外发送的消息  *    Inbox在服务端使用，通过InboxMessage封装接收的消息。  *    outbox内部有messgaes列表存放消息，通过drainOutbox方法循环取出消息并调用sendWith方法处理  *  */private val outboxes = new ConcurrentHashMap[RpcAddress, Outbox]()