概要
上一篇文章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注册和心跳机制就介绍完了。
