1.        case RegisterApplication(description, driver) =>
2.       //待办事项：防止driver程序重复注册
3.      //Master处于STANDBY(备用)状态，不作处理
4.       if (state == RecoveryState.STANDBY) {
5.         //忽略，不发送响应
6.       } else {
7.         logInfo("Registering app " + description.name)
8.     //由description描述，构建ApplicationInfo
9.         val app = createApplication(description, driver)
10.        registerApplication(app)
11.        logInfo("Registered app " + description.name + " with ID " + app.id)
12.     //在持久化引擎中加入application
13.        persistenceEngine.addApplication(app)
14.        driver.send(RegisteredApplication(app.id, self))
15.  //调用schedule方法，在worker节点上启动Executor
16.        schedule()
17.      }

1.  private def createApplication(desc: ApplicationDescription, driver:
        RpcEndpointRef):
2.        ApplicationInfo = {
3.   //ApplicationInfo创建时间
4.      val now = System.currentTimeMillis()
5.      val date = new Date(now)
6.    //由date生成application id
7.      val appId = newApplicationId(date)
8.   //创建ApplicationInfo
9.      new ApplicationInfo(now, appId, desc, date, driver, defaultCores)
10.   }

1. val appId = "app-%s-%04d".format(createDateFormat.format(submitDate),
   nextAppNumber)

1.    private def registerApplication(app: ApplicationInfo): Unit = {
2.    //Driver的地址，用于Master和Driver通信
3.      val appAddress = app.driver.address
4.    //如果addressToApp中已经有了该Driver地址，说明该Driver已经注册过了，直接
      //return
5.
6.      if (addressToApp.contains(appAddress)) {
7.        logInfo("Attempted to re-register application at same address: " +
          appAddress)
8.        return
9.      }
10.  //向度量系统注册
11.     applicationMetricsSystem.registerSource(app.appSource)
12.  //apps是一个HashSet，保存数据不能重复，向HashSet中加入app
13.     apps += app
14. //idToApp是一个HashMap，该HashMap用于保存id和app的对应关系
15.     idToApp(app.id) = app
16. //endpointToApp是一个HashMap，该HashMap用于保存driver和app的对应关系
17.     endpointToApp(app.driver) = app
18. //addressToApp是一个HashMap，记录app Driver的地址和app的对应关系
19.     addressToApp(appAddress) = app
20. /waitingApps是一个数组，记录等待调度的app记录
21.     waitingApps += app
22.     if (reverseProxy) {
23.       webUi.addProxyTargets(app.id, app.desc.appUiUrl)
24.     }
25.   }

1.  private def schedule(): Unit = {
2.  .......
3.       launchDriver(worker, driver)
4.      .......
5.      startExecutorsOnWorkers()
6.    }

1. private def allocateWorkerResourceToExecutors(
2.    ......
3.       launchExecutor(worker, exec)
4.    .......

1.      private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc):
        Unit = {
2.      logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
3.  //向WorkerInfo中加入exec这个描述Executor的ExecutorDesc对象
4.      worker.addExecutor(exec)
5.  //向worker发送LaunchExecutor消息，加载Executor消息中携带了masterUrl地址、
    //application id、Executor id、Executor描述desc、Executor核的个数、Executor
    //分配的内存大小
6.
7.      worker.endpoint.send(LaunchExecutor(masterUrl,
8.        exec.application.id, exec.id, exec.application.desc, exec.cores,
          exec.memory))
9.    //向Driver发回ExecutorAdded消息，消息携带worker的id号、worker的host和
      //port、分配的核的个数和内存大小
10.     exec.application.driver.send(
11.       ExecutorAdded(exec.id,       worker.id,     worker.hostPort,      exec.cores,
          exec.memory))
12.   }

1.   private[worker] def start() {
2.       //创建线程
3.      workerThread = new Thread("ExecutorRunner for " + fullId) {
4.       //线程run方法中调用fetchAndRunExecutor
5.        override def run() { fetchAndRunExecutor() }
6.      }
7.       //启动线程
8.      workerThread.start()
9.
10.      //终止回调函数，用于杀死进程
11.     shutdownHook = ShutdownHookManager.addShutdownHook { () =>
12.       //这是可能的，调用fetchAndRunExecutor       之前，state  将是 ExecutorState.
          //RUNNING。在这种情况下，我们应该设置“状态”为“失败”
13.       if (state == ExecutorState.RUNNING) {
14.         state = ExecutorState.FAILED
15.       }
16.       killProcess(Some("Worker shutting down")) }
17.   }

1.      override def onStart() {
2.  .......
3.        driver = Some(ref)
4.   //向driver发送ask请求，等待driver回应
5.        ref.ask[Boolean](RegisterExecutor(executorId, self, hostname,
          cores, extractLogUrls))
6.    ......

1.     override def receiveAndReply(context: RpcCallContext):
       PartialFunction[Any, Unit] = {
2.
3.       case RegisterExecutor(executorId, executorRef, hostname, cores,
         logUrls) =>
4.  .......
5.           executorRef.send(RegisteredExecutor)
6.   ......

1.   override def receive: PartialFunction[Any, Unit] = {
2.      case RegisteredExecutor =>
3.        logInfo("Successfully registered with driver")
4.        try {
5.    //收到RegisteredExecutor消息，立即创建Executor
6.          executor = new Executor(executorId, hostname, env, userClassPath,
            isLocal = false)
7.        } catch {
8.          case NonFatal(e) =>
9.            exitExecutor(1, "Unable to create executor due to " + e.getMessage, e)
10.       }

1.   override def receive: PartialFunction[Any, Unit] = {
2.      case StatusUpdate(executorId, taskId, state, data) =>
3.        scheduler.statusUpdate(taskId, state, data.value)
4.        if (TaskState.isFinished(state)) {
5.          executorDataMap.get(executorId) match {
6.            case Some(executorInfo) =>
7.              executorInfo.freeCores += scheduler.CPUS_PER_TASK
8.              makeOffers(executorId)
9.            case None =>
10.             //忽略更新，因为我们不知道Executor
11.             logWarning(s"Ignored task status update ($taskId state $state)"+
12.               s"from unknown executor with ID $executorId")
13.         }
14.       }