【原】Spark Rpc通信源码分析

时间:2022-11-05 19:28:19

Spark 1.6+推出了以RPCEnv、RPCEndpoint、RPCEndpointRef为核心的新型架构下的RPC通信方式。其具体实现有Akka和Netty两种方式,Akka是基于Scala的Actor的分布式消息通信系统,Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。

Rpc Environment(RpcEnv)是一个RpcEndpoints用于处理消息的环境,它管理着整个RpcEndpoints的声明周期:(1)根据name或uri注册endpoints(2)管理各种消息的处理(3)停止endpoints。RpcEnv必须通过工厂类RpcEnvFactory创建。

RpcEndpoint需要注册到RpcEnv,RpcEnv处理从RpcEndpointRef或远程节点发送过来的消息,然后把响应消息给RpcEndpoint。对于Rpc捕获到的异常消息,RpcEnv将会用RpcCallContext.sendFailure将失败消息发送给发送者,或者将没有发送者、‘NotSerializableException’等记录到日志中。同时,RpcEnv也提供了根据name或uri获取RpcEndpointRef的方法。

【原】Spark Rpc通信源码分析

Rpc、RpcEndpoint、RpcEndpointRef三者关系

1.RpcEnv源码分析

1.根据RpcEndpoint返回RpcEndpointRef,具体实现在RpcEndpoint.self方法中,如果RpcEndpointRef不存在,将返回null

private[rpc] def endpointRef(endpoint: RpcEndpoint): RpcEndpointRef

2.根据RpcEndpoint的name注册到RpcEnv中并返回它的一个引用RpcEndpointRef

def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef

3.获取RpcEndpointRef的方法

(1)通过url获取RpcEndpointRef

//通过url异步获取RpcEndpointRef

def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef]

//通过url同步获取RpcEndpointRef,这是一个阻塞操作

def setupEndpointRefByURI(uri: String): RpcEndpointRef = {

defaultLookupTimeout.awaitResult(asyncSetupEndpointRefByURI(uri))}

(2)根据systemName、address、endpointName获取RpcEndpointRef,其实是将三者拼接为uri,根据uri获取

//异步获取

def asyncSetupEndpointRef(

systemName: String, address: RpcAddress, endpointName: String): Future[RpcEndpointRef] = {

asyncSetupEndpointRefByURI(uriOf(systemName, address, endpointName))}

//同步获取

def setupEndpointRef(

systemName: String, address: RpcAddress, endpointName: String): RpcEndpointRef = {

setupEndpointRefByURI(uriOf(systemName, address, endpointName))

}

4.根据RpcEndpointRef停止RpcEndpoint

def stop(endpoint: RpcEndpointRef): Unit

5.等待直到RpcEnv退出

def awaitTermination(): Unit

6.RpcEndpointRef需要RpcEnv来反序列化,所以当反序列化RpcEndpointRefs的任何object时,应该通过该方法来操作

def deserializeT: T

2.RpcEndpoint源码分析

RpcEndpoint定义了由消息触发的一些函数,onStart, receive and onStop的调用是顺序发生的。它的声明周期是constructor -> onStart -> receive* -> onStop。注意,receive能并发操作,如果你想要receive是线程安全的,请使用ThreadSafeRpcEndpoint,如果RpcEndpoint抛出错误,它的onError方法将会触发。它有51个实现子类,我们比较熟悉的是Master、Worker、ClientEndpoint等。

1.启动RpcEndpoint处理任何消息

def onStart(): Unit = {}

2.停止RpcEndpoint

def onStop(): Unit = {}

3.处理RpcEndpointRef.send或RpcCallContext.reply方法,如果收到不匹配的消息,将抛出SparkException

def receive: PartialFunction[Any, Unit] = {

case _ => throw new SparkException(self + " does not implement 'receive'")}

4.处理RpcEndpointRef.ask方法,如果不匹配消息,将抛出SparkException

def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {

case _ => context.sendFailure(new SparkException(self + " won't reply anything"))}

5.当处理消息发生异常时

def onError(cause: Throwable): Unit = {

throw cause}

6.当远程地址连接到当前的节点地址时触发

def onConnected(remoteAddress: RpcAddress): Unit = {

}

7.当远程地址连接断开时触发

def onDisconnected(remoteAddress: RpcAddress): Unit = {

}

8.当远程地址和当前节点的连接发生网络异常时触发

def onNetworkError(cause: Throwable, remoteAddress: RpcAddress): Unit = {

// By default, do nothing.

}

3.RpcEndpointRef源码分析

RpcEndpointRef是RpcEndpoint的一个远程引用,是线程安全的。它有两个实现子类:即AkkaRpcEndpointRef和NettyRpcEndpointRef。

1.发送单方面的异步消息

def send(message: Any): Unit

2.发送一个消息给RpcEndpoint.receiveAndReply并返回一个Future在指定的时间内接受响应,本方法值请求一次

def askT: ClassTag: Future[T]

3.发送消息给RpcEndpoint并在默认的超时内得到结果,否则抛出SparkException,注意,本方法是一个阻塞操作可能消耗时间,所以不要早消息循环中调用它

def askWithRetryT: ClassTag: T = askWithRetry(message, defaultAskTimeout)

最后,画图说明一下两者的消息传递的过程,RpcEndpointRef作为消息的主动者,RpcEndpoint作为消息的被动者

【原】Spark Rpc通信源码分析

RpcEndpoint、RpcEndpointRef