Spark 中的 RPC

Spark 是一个 通用的分布式计算系统，既然是分布式的，必然存在不少节点之间的通讯，那么 Spark 不一样组件之间就会经过 RPC（Remote Procedure Call）进行点对点通讯。

Spark 的 RPC 主要在两个模块中：

1，spark-core 中，主要承载了更好的封装 server 和 client 的做用，以及和 scala 语言的融合，它依赖 spark-network-common 模块；

2，spark-network-common 中，该模块是 java 语言写的，最新版本是基于 Netty 开发的；

Spark 早期版本中使用 Netty 通讯框架作大块数据的传输，使用 Akka 用做 RPC 通讯。自 Spark2.0 以后已经把 Akka 框架玻璃出去了（详见SPARK-5293），是由于不少用户会使用 Akka 作消息传递，会与 Spark 内嵌的版本产生冲突。在 Spark2.0 以后，基于底层的 spark-network-commen 模块实现了一个相似 Akka Actor 消息传递模式的 scala 模块，封装在 spark-core 中。

看一张 UML 图，图内展现了 Spark RPC 模块内的类的关系，白色的是 spark-core 中的类，黄色的 spark-common 中的类：

整个 Spark 的 RPC 模块大概有几个主要的类构成：

1，RpcEndPonit 和 RpcCallContext，RpcEndPoint 是一个能够相应请求的服务，相似于 Akka 中的 Actor。其中有 receive 方法用来接收客户端发送过来的信息，也有 receiveAndReply 方法用来接收并应答，应答经过 RpcContext 回调。能够看下面代码：

def receive: PartialFunction[Any, Unit] = {
    case _ => throw new RpcException(self + " does not implement 'receive'")
}

def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
    case _ => context.sendFailure(new RpcException(self + " won't reply anything"))
}
复制代码

2，RpcEndpointRef，相似于 Akka 中的 ActorRef，是 RpcEndPoint 的引用，持有远程 RpcEndPoint 的地址名称等，提供了 send 方法和 ask 方法用于发送请求。能够看看 RpcEndPoint 内部的成员变量和方法：

/** * return the address for the [[RpcEndpointRef]] */
  def address: RpcAddress

  def name: String

  /** * Sends a one-way asynchronous message. Fire-and-forget semantics. */
  def send(message: Any): Unit

  /** * Send a message to the corresponding [[RpcEndpoint.receiveAndReply)]] and return a [[Future]] to * receive the reply within the specified timeout. * * This method only sends the message once and never retries. */
  def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]
复制代码

3，RpcEnv 和 NettyRpcEnv

RpcEnv 相似于 ActorSystem，服务端和客户端均可以使用它来作通讯。

对于 server 端来讲，RpcEnv 是 RpcEndpoint 的运行环境，负责 RpcEndPoint 的生命周期管理，解析 Tcp 层的数据包以及反序列化数据封装成 RpcMessage，而后根据路由传送到对应的 Endpoint；

对于 client 端来讲，能够经过 RpcEnv 获取 RpcEndpoint 的引用，也就是 RpcEndpointRef，而后经过 RpcEndpointRef 与对应的 Endpoint 通讯。

RpcEnv 中有两个最经常使用的方法：

// 注册endpoint，必须指定名称，客户端路由就靠这个名称来找endpoint
def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef 

// 拿到一个endpoint的引用
def setupEndpointRef(address: RpcAddress, endpointName: String): RpcEndpointRef
复制代码

NettyRpcEnv 是 spark-core 和 spark-network-common 的桥梁，内部 leverage 底层提供的通讯能力，同事包装了一个类 Actor 的语义。

4，Dispatcher ，NettyRpcEnv 中包含 Dispatcher，主要针对服务端，帮助路由到指定的 RpcEndPoint，并调用起业务逻辑。

参考：

1，blog.csdn.net/justlpf/art…

2，zhuanlan.zhihu.com/p/28893155