/** * An RPC environment. [[RpcEndpoint]]s need to register itself with a name to [[RpcEnv]] to * receives messages. Then [[RpcEnv]] will process messages sent from [[RpcEndpointRef]] or remote * nodes, and deliver them to corresponding [[RpcEndpoint]]s. For uncaught exceptions caught by * [[RpcEnv]], [[RpcEnv]] will use [[RpcCallContext.sendFailure]] to send exceptions back to the * sender, or logging them if no such sender or `NotSerializableException`. * * [[RpcEnv]] also provides some methods to retrieve [[RpcEndpointRef]]s given name or uri. */

    private def runDriver(): Unit = {        addAmIpFilter(None)
        /*        这里，调用startUserApplication方法来执行用户的class，也就是我们的jar包，        invoke我们的main方法，从而启动了sparkContext，内部启动一系列的scheduler以及        backend，以及taskscheduler等等等等core的内容，其他篇章会详细讲解        */          userClassThread = startUserApplication()
        // This a bit hacky, but we need to wait until the spark.driver.port property has        // been set by the Thread executing the user class.        logInfo("Waiting for spark context initialization...")        val totalWaitTime = sparkConf.get(AM_MAX_WAIT_TIME)        try {
          /*          这里，阻塞的等待SparkContext从Driver线程中返回回来          */          val sc = ThreadUtils.awaitResult(sparkContextPromise.future,            Duration(totalWaitTime, TimeUnit.MILLISECONDS))          if (sc != null) {            rpcEnv = sc.env.rpcEnv
            val userConf = sc.getConf            val host = userConf.get("spark.driver.host")            val port = userConf.get("spark.driver.port").toInt            registerAM(host, port, userConf, sc.ui.map(_.webUrl))
            /*            **这里，上演了好戏，通过NettyRpcEnv的setupEndpointRef方法来获取到driverRef            这个里面其实是去ask一下Driver你在吗？是否存在这个Driver的服务，如果存在，则            返回OK，构建出Driver的Ref**            */            val driverRef = rpcEnv.setupEndpointRef(              RpcAddress(host, port),              YarnSchedulerBackend.ENDPOINT_NAME)            createAllocator(driverRef, userConf)          } else {            // Sanity check; should never happen in normal operation, since sc should only be null            // if the user app did not create a SparkContext.            throw new IllegalStateException("User did not initialize spark context!")          }          resumeDriver()          userClassThread.join()        } catch {          case e: SparkException if e.getCause().isInstanceOf[TimeoutException] =>            logError(              s"SparkContext did not initialize after waiting for $totalWaitTime ms. " +               "Please check earlier log output for errors. Failing the application.")            finish(FinalApplicationStatus.FAILED,              ApplicationMaster.EXIT_SC_NOT_INITED,              "Timed out waiting for SparkContext.")        } finally {          resumeDriver()        }      }

    import scala.concurrent.ExecutionContext.Implicits.global    import scala.concurrent.Future
    /**     * 解读Future的基础     */    object DocFutureTest {
      def apply(): Unit = {        println("I am DocFutureTest")      }
      def main(args: Array[String]): Unit = {
        val sleeping = 3000;        val main_thread = Thread.currentThread().getName;                /*        定义另一个线程发生的事件        这个事件相当于java中的如下的代码块：        从整体的间接性上看，scala的更为优雅一些，直接一个Future可以包裹住左右需要处理的内容        后续如果需要进行异常处理的话还可以根据Success和Failture进行模式匹配        public class JavaThreading {                    public static void main(String[] args) throws InterruptedException {                new Thread(                        () -> System.out.println("这是一条发生在另一个叫做叫做" + Thread.currentThread().getName() + " 线程的故事")                ).start();                        System.out.println(Thread.currentThread().getName());                Thread.sleep(3000);            }        }        */        var future_run = Future {          Thread.sleep(1000)          println("这是一条发生在另一个叫做叫做" + Thread.currentThread().getName +" 线程的故事")        }
        // 主线程休息3000ms        // 如果不休息的话，main线程会先停止，导致上面的Future定义的thread还没有被执行到就结束了        Thread.sleep(sleeping)        println(s"$main_thread 线程休息 $sleeping 毫秒")
      }
    }

    case class ExceptionError(error: String) extends Exception(error)
      def main(args: Array[String]): Unit = {
        val sleeping = 3000;        val main_thread = Thread.currentThread().getName;
        // 定义另一个线程发生的事件        var future_run = Future {          Thread.sleep(1000)          prntln("这是一条发生在另一个叫做叫做" + Thread.currentThread().getName + " 线程的故事")        // 如果需要onFailure的话 则释放此句        // throw ExceptionError("error") 
        future_run onFailure {          case t => println("exception " + t.getMessage)        }
        future_run onSuccess {          case _ => println("success")        }

    import scala.concurrent.Promise    import scala.util.{Failure, Success}
    object PromiseTest {      def main(args: Array[String]): Unit = {        import scala.concurrent.ExecutionContext.Implicits.global        val promise = Promise[String]        promise.future.onComplete(v => println("onComplete " + v))        promise.future.map(str => println("future: " + str + " ==> " + Thread.currentThread().getName))        promise.future.failed.foreach(e => println(e + " ==> " + Thread.currentThread().getName))        Thread.sleep(3000)      }
    }

    import scala.concurrent.Promise    import scala.util.{Failure, Success}
    object PromiseTest {      def main(args: Array[String]): Unit = {        import scala.concurrent.ExecutionContext.Implicits.global        val promise = Promise[String]        promise.future.onComplete(v => println("onComplete " + v))        promise.future.map(str => println("future: " + str + " ==> " + Thread.currentThread().getName))        promise.future.failed.foreach(e => println(e + " ==> " + Thread.currentThread().getName))        **promise.trySuccess("try success "  + " --> " + Thread.currentThread().getName)**        Thread.sleep(3000)      }
    }

private[netty] def ask[T: ClassTag](message: RequestMessage, timeout: RpcTimeout): Future[T] = {    // 定义了一个Any的promise    val promise = Promise[Any]()    val remoteAddr = message.receiver.address
    def onFailure(e: Throwable): Unit = {      if (!promise.tryFailure(e)) {        e match {          case e : RpcEnvStoppedException => logDebug (s"Ignored failure: $e")          case _ => logWarning(s"Ignored failure: $e")        }      }    }
    /*     这里声明的onSuccess会被填充到RpcResponseCallback的onSuccess中，这个    RpcResponseCallback就是上面【图9】中的listener，当我们从Server端获取到response后    注意，获取的不是RpcFailure类型的response，则都会进入到【图9】的    else if (message instanceof RpcResponse) { 分支中
    */    def onSuccess(reply: Any): Unit = reply match {      case RpcFailure(e) => onFailure(e)      case rpcReply =>        /*        当返回的response是OK的没有问题后，onSuccess被callback，这里promise的trySuccess也        进行call操作，这里就是上面所说的，为了一个promise铺设了一条future，从而可以执行        这个Future的线程了        */        if (!promise.trySuccess(rpcReply)) {          logWarning(s"Ignored message: $reply")        }    }
    try {      if (remoteAddr == address) {        val p = Promise[Any]()        p.future.onComplete {          case Success(response) => onSuccess(response)          case Failure(e) => onFailure(e)        }(ThreadUtils.sameThread)        dispatcher.postLocalMessage(message, p)      } else {        val rpcMessage = RpcOutboxMessage(message.serialize(this),          onFailure,          (client, response) => **onSuccess**(deserialize[Any](client, response)))        postToOutbox(message.receiver, rpcMessage)        /*        如果是callback了Failure，则这里会被执行        */        promise.future.failed.foreach {          case _: TimeoutException => rpcMessage.onTimeout()          case _ =>        }(ThreadUtils.sameThread)      }
      val timeoutCancelable = timeoutScheduler.schedule(new Runnable {        override def run(): Unit = {          onFailure(new TimeoutException(s"Cannot receive any reply from ${remoteAddr} " +            s"in ${timeout.duration}"))        }      }, timeout.duration.toNanos, TimeUnit.NANOSECONDS)
      /*      当promise的future执行后，会调用这里的onComplete方法      */      promise.future.onComplete { v =>        timeoutCancelable.cancel(true)      }(ThreadUtils.sameThread)    } catch {      case NonFatal(e) =>        onFailure(e)    }
    /*    利用RpcTimeout中的addMessageIfTimeout的偏函数再去模式匹配一下产生的Throwable内容    如果是RpcTimeoutException 则 直接throw这个ex    如果是TimeoutException 则包装成RpcTimeoutException后再throw出去    */    promise.future.mapTo[T].recover(timeout.addMessageIfTimeout)(ThreadUtils.sameThread)  }