关于Spark的RPC机制

一直以来，基于 Akka 实现的 RPC 通信框架是 Spark 引以为豪的主要特性，也是与 Hadoop 等分布式计算框架对比过程中一大亮点，但是时代和技术都在演化，从 Spark1.3.1 版本开始，为了解决大块数据（如 Shuffle ）的传输问题， Spark 引入了 Netty 通信框架，到了 1.6.0 版本， Netty 完成取代了 Akka ，承担 Spark 内部所有的 RPC 通信以及数据流传输。

Spark RPC 中最为重要的三个抽象（“三剑客”）为：RpcEnv 、 RpcEndpoint 、 RpcEndpointRef ，这样做的好处有：

l 对上层的 API 来说，屏蔽了底层的具体实现，使用方便

l 可以通过不同的实现来完成指定的功能，方便扩展

l 促进了底层实现层的良性竞争， Spark 1.6.3 中默认使用了 Netty 作为底层的实现，但 Akka 的依赖仍然存在；而 Spark 2.1.0 中的底层实现只有 Netty ，这样用户可以方便的使用不同版本的 Akka 或者将来某种更好的底层实现。

Send a message locally

通过 Spark 源码中的一个 Test （ RpcEnvSuite.scala ）来分析一下发送本地消息的具体流程，源码如下（对源码做了一些修改）：

test( "send a message locally" ) {

@volatile var message: String = null

val rpcEndpointRef = env.setupEndpoint( "send-locally" , new RpcEndpoint {

override val rpcEnv = env

override def receive = {

//case msg: String => message = msg

case msg: String => println(message) // 我们直接将接收到的消息打印出来

}

})

rpcEndpointRef.send( "hello" )

// 下面是原来的代码

//eventually(timeout(5 seconds), interval(10 millis)) {

// assert("hello" === message)

//}

}

首先是 RpcEndpoint 创建并注册的流程：

· 1 、 创建 RpcEndpoint ，并初始化 rpcEnv 的引用（ RpcEnv 已经创建好，底层实际上是实例化了一个 NettyRpcEnv ，而 NettyRpcEnv 是通过工厂方法 NettyRpcEnvFactory 创建的）

· 2 、 实例化 RpcEndpoint 之后需要向 RpcEnv 注册该 RpcEndpoint ，底层实现是向 NettyRpcEnv 进行注册，而实际上是通过调用 Dispatcher 的 registerRpcEndpoint 方法向 Dispatcher 进行注册

· 3 、 具体的注册就是向 endpoints 、 endpointRefs 、 receivers 中插入记录：而 receivers 中插入的信息会被 Dispatcher 中的线程池中的线程执行：会将记录 take 出来然后调用 Inbox 的 process 方法通过模式匹配的方法进行处理，注册的时候通过匹配到 OnStart 类型的 message ，去执行 RpcEndpoint 的 onStart 方法（例如 Master 、 Worker 注册时，就要执行各自的 onStart 方法），本例中未做任何操作

· 4 、 注册完成后返回 RpcEndpointRef ，我们通过 RpcEndpointRef 就可以向其代表的 RpcEndpoint 发送消息

下面就是通过 RpcEndpointRef 向其代表的 RpcEndpoint 发送消息的具体流程：

· 1 、 2 、调用 RpcEndpointRef 的 send 方法，底层实现是调用 Netty 的 NettyRpcEndpointRef 的 send 方法，而实际上又是调用的 NettyRpcEnv 的 send 方法，发送的消息使用 RequestMessage 进行封装：

nettyEnv.send( RequestMessage (nettyEnv.address, this , message))

· 3 、 4 、 NettyRpcEnv 的 send 方法首先会根据 RpcAddress 判断是本地还是远程调用，此处是同一个 RpcEnv ，所以是本地调用，即调用 Dispatcher 的 postOneWayMessage 方法

· 5 、 postOneWayMessage 方法内部调用 Dispatcher 的 postMessage 方法

· 6 、 postMessage 会向具体的 RpcEndpoint 发送消息，首先通过 endpointName 从 endpoints 中获得注册时的 EndpointData ，如果不为空就执行 EndpointData 中 Inbox 的 post(message) 方法，向 Inbox 的 mesages 中插入一条 InboxMessage ，同时向 receivers 中插入一条记录，此处将 Inbox 单独画出来是为了方便大家理解

· 7 、 Dispatcher 中的线程池会拿出一条线程用来循环 receivers 中的消息，首先使用 take 方法获得 receivers 中的一条记录，然后调用 Inbox 的 process 方法来执行这条记录，而 process 将 messages 中的一条 InboxMessage （第 6 步中插入的）拿出来进行处理，具体的处理方法就是通过模式匹配的方法，匹配到消息的类型（此处是 OneWayMessage ），然后来执行 RpcEndpoint 中对应的 receive 方法，在此例中我们只打印出这条消息（步骤 8 ）

至此，一个简单的发送本地消息的流程执行完成。