概要

上一篇文章Spark RPC之Master实现介绍了standalone模式下Master端的实现，接着我们看下Worker端的实现，以及Worker如何向Master注册信息及发送心跳。

Worker

查看Worker，Worker也是RpcEndpoint的子类，所以接下来查看RpcEndpoint生命周期的四个方法: onStart -> receive(receiveAndReply)* -> onStop。

onStart

  override def onStart() {
    registerWithMaster()
  }

本篇的第一个重要部分，向Master注册，查看registerWithMaster方法

如图中注释，调用tryRegisterAllMasters注册，查看tryRegisterAllMasters方法

private def tryRegisterAllMasters(): Array[JFuture[_]] = {
    masterRpcAddresses.map { masterAddress =>
      registerMasterThreadPool.submit(new Runnable {
        override def run(): Unit = {
            val masterEndpoint = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
            sendRegisterMessageToMaster(masterEndpoint)
        }
      })
    }
  }

val masterEndpoint = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME) masterEndpoint 持有 返回的 RpcEndpointRef 类型对象的 引用，查看 sendRegisterMessageToMaster 方法：

  private def sendRegisterMessageToMaster(masterEndpoint: RpcEndpointRef): Unit = {
    masterEndpoint.send(RegisterWorker(...))
  }

调用了 RpcEndpointRef 的 send 方法，把 RegisterWorker 类型消息发送到 Master，Master 收到 Worker 的 RegisterWorker 消息：

override def receive: PartialFunction[Any, Unit] = {
  case RegisterWorker(
        val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
          workerRef, workerWebUiUrl)
        if (registerWorker(worker)) {
          persistenceEngine.addWorker(worker)
          //发送 RegisteredWorker 消息给 Worker
          workerRef.send(RegisteredWorker(self, masterWebUiUrl, masterAddress))
          schedule()
        }
}

Worker 在 receive 方法中接受消息：

override def receive: PartialFunction[Any, Unit] = synchronized {
    case msg: RegisterWorkerResponse =>
      handleRegisterResponse(msg)
}

Worker 收到 RegisterWorkerResponse 类型消息，则调用 handleRegisterResponse 方法：

  private def handleRegisterResponse(msg: RegisterWorkerResponse): Unit = synchronized {
    msg match {
      case RegisteredWorker(masterRef, masterWebUiUrl, masterAddress) =>
        registered = true
        
        forwordMessageScheduler.scheduleAtFixedRate(new Runnable {
          override def run(): Unit = Utils.tryLogNonFatalError {
            self.send(SendHeartbeat)
          }
        }, 0, HEARTBEAT_MILLIS, TimeUnit.MILLISECONDS)

      case RegisterWorkerFailed(message) =>

      case MasterInStandby =>
    }
  }

返回的消息类型有三种RegisteredWorker、RegisterWorkerFailed和MasterInStandby，这三种消息都继承自：RegisterWorkerResponse，所以可以在 receive 中匹配成功。
在 handleRegisterResponse 方法中，匹配到 Master 发送的 RegisteredWorker 消息，并最终调用 self 的 send(SendHeartbeat) 发送给你自己，Worker自己接收并处理：

    case SendHeartbeat =>
      if (connected) { sendToMaster(Heartbeat(workerId, self)) }

sendToMaster 方法：

  private def sendToMaster(message: Any): Unit = {
    master match {
      case Some(masterRef) => masterRef.send(message)
    }
  }

还是调用 RpcEndpointRef send方法，发送 Heartbeat 消息 到 Master，Master receive 中接收：

    case Heartbeat(workerId, worker) =>
      idToWorker.get(workerId) match {
        case Some(workerInfo) =>
          workerInfo.lastHeartbeat = System.currentTimeMillis()
      }

最后，Master每60s查看Worker连接情况，Worker端每15s发送一次心跳，如下

receive

SendHeartbeat是上面刚讲过的，其他限于篇幅不再讲述。

receiveAndReply

只处理Worker状态查询。

onStop

相比于Master，多了executors、drivers和shuffleService的关闭。

Main

启动程序和Master几乎一致

总结

上一篇Spark RPC之Master实现讲述了Master是如何接受Worker请求注册的信息和心跳机制，本篇文章讲解了Worker端对应的行为，如下
1. Worker的onStart方法中如何发送注册信息给Master
2. 在注册成功后，Worker在处理Master返回信息时，启动定时任务，每15s发送心跳
完整流程如下

至此，standalone模式下，spark如何使用rpc完成Worker注册和心跳机制就介绍完了。