如何进行Spark底层通信RPC源码分析

共计 6027 个字符，预计需要花费 16 分钟才能阅读完成。

本篇文章给大家分享的是有关如何进行 Spark 底层通信 RPC 源码分析，丸趣 TV 小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着丸趣 TV 小编一起来看看吧。

RPC 通信: 无论是 hadoop2.x 的 Rpc 通信方式还是 Spark2.x 的 Rpc 通信方式，简单通俗的来说就是两个进程之间的远程通信，比如 java 一个 A 项目里面有一个 class A，里面有一个 washA 方法一个 B 项目里面有一个 Class B 类，里面有一个方法是 washB，B 项目通过代理模式以及 java 的反射机制调用到 A 项目里面的 washA, 这种情况下就可以理解为是一个简单的 Rpc 通信方式。

Spark2.x

Spark2.x 使用基于 RPC 的通信方式，去除了 1.x 的 Akka 的实现方式，只保留了 netty 的实现方式，Spark2.x Rpc 提供了上层抽象（RpcEndpoint、RpcEnv、RpcEndPointRef），具体的实现方式只要实现了定义的抽象就可以完成 Rpc 通信，Spark2.x 之后目前版本只保留了 Netty（NettyRpcEnv、NettyRpcEndpointRef）的实现，定义抽象最大的好处相信开发的朋友都很清楚，以后不管提供了什么方式的实现只要实现了 RPCEndpoint,RpcEnv，RpcEndpointRef 就可以完成的通信功能。比如自己写一个自己版本的 Rpc 通信实现。

Spark2.x 的 Rpc 通信方式主要包括一下几个重要方面

RpcEndpoint: 消息通信体，主要是用来接收消息、处理消息，实现了 RpcEndPoint 接口就是一个消息通信体（Master、Work），RpcEndpoint 需要向 RpcEnv 注册

RpcEnv:Rpc 通信的上下文环境，消息发送过来首先经过 RpcEnv 然后路由给对应的 RpcEndPoint，得到 RpcEndPoint

RpcEndPointRef:RpcEndPoint 的引用如果要想某个 RpcEndPoint 发送消息，首先要通过 RpcEnv 得到 RpcEndPoint 的引用

RpcEndPoint 接口 里面的定义如下

val rpcEnv : RpcEnv // 得到 RpcEnv 对象

final def self: RpcEndpointRef = {// 返回一个 RpcEnpointRef 这个方法通常用来自己给自己发送消息

  rpcEnv.endpointRef(this)

  }

def receive: PartialFunction[Any, Unit]// 处理 RpcEndPointRef.send 或者 RpcEndPointRef.reply 方法，该方法不需要进行响应信息

def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit]// 处理 RpcEndPointref.ask 发送的消息，处理完之后需要给调用 ask 的通信端响应消息（reply）

def onError(cause: Throwable)// 处理消息失败的时候会调用此方法

def onConnected(remoteAddress: RpcAddress)// 远程连接的当前节点的时候触发

def onDisconnected(remoteAddress: RpcAddress)// 远程连接断开时候触发

def onNetworkError(cause: Throwable, remoteAddress: RpcAddress)// 远程连接发生网络异常时触发

def onStop()// 停止 RpcEndPoint

def onStart()// 启动 RpcEndPoint，这里不仅仅是网络上说的启动 RpcEndPoint 处理任何消息，onStart 方法里面很多情况下可以写自己的 RpcEndPoint 的一些实现比如启动端口，或者创建目录

但是 RpcEndPoint 只有在 onStart 方法做一些处理之后 才可以接受 RpcEndPointRef 发送的消息

private[spark] trait ThreadSafeRpcEndpoint extends RpcEndpoint// 因为 receive 是并发操作如果要现成安全就是用 threadSafeRpcEndPoint

RpcEndPoint 的生命周期 构造 – onStart– receive — onStop, 注意 onStart 的方法是在调用 setRpcEndPoint 注册之后就会执行任何 RpcEndPoint 的 onStart 方法都是在注册之后执行的

原因后面的源码的提到

RpcEndpointRef: 抽象类

  def address: RpcAddress // 根据主机名端口返回一个 RppAddress

def name: String//name 一个字符串 暂时不知道干嘛的

def send(message: Any): Unit// 向 RpcEndPoint 发送一个消息 不需要返回结果

 def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T]

  def ask[T: ClassTag](message: Any): Future[T] = ask(message, defaultAskTimeout) // 向 RpcEndPoint 发送消息并得到返回结果

def askWithRetry[T: ClassTag](message: Any): T = askWithRetry(message, defaultAskTimeout)// 想 RpcEndPoint 发送消息并在一定时间内返回结果 失败的时候并且进行一定次数的重试

RpcEnv

  private[rpc] def endpointRef(endpoint: RpcEndpoint): RpcEndpointRef// 传入 RpcEndPoint 得到 RpcEndPointref 对象

  def address: RpcAddress// 根据主机名端口返回一个 RppAddress

def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef// 注册 RpcEndPoint 返回对应的 RpcEndPointRef

def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef]// 通过 uri 一步获取 RpcEndPointRef

 def stop(endpoint: RpcEndpointRef): Unit// 停止 RpcEndPoint 根据 RpcEndPointRef

  def shutdown(): Unit// 关闭 RpcEndPoint

 def awaitTermination(): Unit// 等待 RpcEndPoint 退出

object RpcEnv

 def create(

  name: String,

  host: String,

  port: Int,

  conf: SparkConf,

  securityManager: SecurityManager,

  clientMode: Boolean = false): RpcEnv = {

  val config = RpcEnvConfig(conf, name, host, port, securityManager, clientMode)

  new NettyRpcEnvFactory().create(config)

  }

// 通过 RpcEnvFactory.create 创建 RpcEnv 环境

RpcEnvConfig

private[spark] case class RpcEnvConfig(

  conf: SparkConf,

  name: String,

  host: String,

  port: Int,

  securityManager: SecurityManager,

  clientMode: Boolean)

case 类 里面包括 SparkConf，name,host,port 等

NettyRpcEnv NettyRpcEnv 通过 NettyRpcEnvFactory 的 create 方法创建

 val nettyEnv =

  new NettyRpcEnv(sparkConf, javaSerializerInstance, config.host, config.securityManager)// 创建 nettyEnv

 private val dispatcher: Dispatcher = new Dispatcher(this)

Dispatcher 负责 RPC 消息的路由，它能够将消息路由到对应的 RpcEndpoint 进行处理, 同时存放 RpcEndPoint 与 RpcEndPointRef 的映射

NettyStreamManager 负责提供文件服务（文件、JAR 文件、目录）

TransportContext 负责管理网路传输上下文信息：创建 MessageEncoder、MessageDecoder、TransportClientFactory、TransportServer

NettyRpcHandler 负责处理网络 IO 事件，接收 RPC 调用请求，并通过 Dispatcher 派发消息

这里说一下 Dispatcher 该类主要负责 Rpc 消息路由 里面有一个内部累 EndPointData 但是有一个现成安全的 Inbox 这里面存放的时候收到的消息，非常重要后面会做具体分析

private class EndpointData(

  val name: String,

  val endpoint: RpcEndpoint,

  val ref: NettyRpcEndpointRef) {

  val inbox = new Inbox(ref, endpoint)

  }

  private val endpoints = new ConcurrentHashMap[String, EndpointData]// 存放 name- 对应的 EndPoint 的信息

  private val endpointRefs = new ConcurrentHashMap[RpcEndpoint, RpcEndpointRef]// 存放 RpcEndpoint, RpcEndpointRef 的映射关系

  private val receivers = new LinkedBlockingQueue[EndpointData]// 队列下面会有一个现成不断的从里面取出来处理

 def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {

  val addr = RpcEndpointAddress(nettyEnv.address, name)

  val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)

  synchronized {

  if (stopped) {

  throw new IllegalStateException(RpcEnv has been stopped)

  }

  if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {

  throw new IllegalArgumentException(s There is already an RpcEndpoint called $name)

   }

  val data = endpoints.get(name)

  endpointRefs.put(data.endpoint, data.ref)

  receivers.offer(data)  // for the OnStart message

  }

  endpointRef

  }

// 注册 RpcEndPoint 在这里面发生 同时将 data put 到 receivers 

在 NettyRpcEndPoint 里面有一个 threadpool

private val threadpool: ThreadPoolExecutor = {

  val numThreads = nettyEnv.conf.getInt(spark.rpc.netty.dispatcher.numThreads ,

  math.max(2, Runtime.getRuntime.availableProcessors()))

  val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, dispatcher-event-loop)

  for (i – 0 until numThreads) {

  pool.execute(new MessageLoop)

  }

  pool

  }

MessageLoop 是一个实现了 Runnable 的类，里面的 run 方法里面不断从 receivers 取出来进行处理

重要代码   data.inbox.process(Dispatcher.this)

这个里面有一个非常重要的点就是 什么时候调用 onStart 的方法因为 receivers 里面存放的是 EndPoint 的信息 同时创建 EndPointData 对象

进入 Inbox 里面看一下

  inbox =   // Give this an alias so we can use it more clearly in closures.

  @GuardedBy(this)

  protected val messages = new java.util.LinkedList[InboxMessage]()

 inbox.synchronized {

  messages.add(OnStart)

  }

创建这个类的时候会有一个 messagelinkedList 的 list 集合 在创建这个结合之后 就会将 onStart 方法添加到里面，并且是现成安全的

然后 process 方法里面会不断的拿到集合的数据来进行对应的操作

 case OnStart =

  endpoint.onStart()

  if (!endpoint.isInstanceOf[ThreadSafeRpcEndpoint]) {

  inbox.synchronized {

  if (!stopped) {

  enableConcurrent = true

  }

  }

  }

这个时候就会调用 onStart 方法

这个时候相当于 RpcEndPoint 可以接受消息并且处理了

Spark Rpc 通信方式 分为本地消息和远程消息，本地消息相当于调用的方法直接存放到 Index（中文收件箱），远程消息需要走 NettyRpcHandler

以上就是如何进行 Spark 底层通信 RPC 源码分析，丸趣 TV 小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注丸趣 TV 行业资讯频道。