Spark的Rpct模块的学习

Spark的Rpc模块是1.x重构出来可，以前的代码中大量使用了akka的类，为了把akka从项目的依赖中移除，所有添加了该模块。先看下该模块的几个主要的类

 

使用EA把该模块所有的类都添加进来了

 

要看懂该模块还是要先了解akka,  akka有Actor和ActorRef两个类，一个用于接收消息，一个用于发送消息。正好对应该模块的RpcEndpoint和RpcEndpointRef两个类。

下面大致介绍下这几个类，附带一些scala的特性

1：RpcAddress 

该类是一个case class, 用来表示主机名和端口号 ，  case class也可以添加方法，以前还以为不可以呢

它的伴生对象用于从URI,String等构造一个RpcAddress对象

 

2：RpcTimeout

表示一个超时时间，话说该类的职责有点乱，竟然还有下面的一个方法

1. def awaitResult[T](awaitable:Awaitable[T]): T ={
   try{
   Await.result(awaitable, duration)
   }catch addMessageIfTimeout
   }

在规定时间内返回对象， Await是scala并发库中的一个对象，result在duration时间片内返回Awaitable的执行结果，ready表示duration时间片内Awaitable的状态变成complete,两个方法都是阻塞的，Awaitable相当java中的future，当然scala也有future类，正是继承该类。

它的伴生对象主要是配置文件中获取时间值然后生成该对象

 

3：RpcEnvFactory

该对象用于创建一个RpcEnv，在RpcEnv中可以看到如何使用该方法

2. private def getRpcEnvFactory(conf:SparkConf):RpcEnvFactory={
   // Add more RpcEnv implementations here
   val rpcEnvNames =Map("akka"->"org.apache.spark.rpc.akka.AkkaRpcEnvFactory")
   val rpcEnvName = conf.get("spark.rpc","akka")
   val rpcEnvFactoryClassName = rpcEnvNames.getOrElse(rpcEnvName.toLowerCase, rpcEnvName)
   Utils.classForName(rpcEnvFactoryClassName).newInstance().asInstanceOf[RpcEnvFactory]
   }

目前spark.rpc只有akka的实现,如果觉得akka的性能不好也可以自己实现一个Rpc框架。

 

 

4： RpcEnv

 注解：这是一个RPC环境，所有的RpcEndpont需要注册到该对象中用于接收消息，注册时需要指定一个name, RpcEnv将会处理从RpcEndpontRef和远程节点发送过来的消息（接口里面看不到这块逻辑），然后发送给相应的Endpoint处理，对于接收到的异常使用RpcCallContext来处理。

 

看RpcEnv像akka中的ActorSystem对象，所有的actor和acotorred都属于它，同时有一个根地址，所有RpcEnv有注册RpcEndpoint的方法，也有一个address返回根地址的方法，RpcEnv有几个方法用于获取RpcEndpointRef ,  这里说下Endpoint注册名会成为RpcEndpoint的地址,可以看uriof方法，还有停止和关闭的方法。

RpcEnv的deserialize不明白具体用法，RpcEndpiontRef只能使用RpcEnv解码，当包含有RpcEndpointRef的对象解码时，解码代码将会被方法包装

 

5：RpcEnvConfig

用于构建RpcEnv的配置对象，一个RpcEnv需要host,port,name,附带sparkconf,securitymananger

host,port,name构造结构入下  akka://host:port/name  大致如此

 

 

6：RpcEndpoint

进程间调用的一个端点，当一个消息到来时，方法调用顺序为  onStart, receive, onStop

它的生命周期为constructor -> onStart -> receive* -> onStop  .当然还有一些其他方法，都是间触发方法

 

7：RpcEndpointRef

一个远程RpcEndpoint的引用，通过它可以给远程RpcEndpoint发送消息，可以是同步可以是异步，它映射一个地址 ，

 

 

来自为知笔记(Wiz)