Skip to content

Latest commit

 

History

History
44 lines (32 loc) · 2.05 KB

0043.md

File metadata and controls

44 lines (32 loc) · 2.05 KB

第43课：Spark 1.6 RPC内幕解密：运行机制、源码详解、Netty与Akka等

标签： sparkIMF

##一：Spark 1.6 RPC解析

  1. Spark 1.6推出了以RpcEnv、RpcEndpoint、RpcEndpointRef为核心的新型架构下的RPC通信方式，就目前的实现而言，其底层依旧是Akka；
  2. Akka是基于Actor的分布式消息通信系统，而在Spark 1.6中封装了Akka，提供更高层的Rpc实现，目的是移除对Akka的依赖，为扩展和自定义Rpc打下基础；

##二：RpcEnv解析

  1. RpcEnv是RPC的环境（相当于Akka中的ActorSystem），所有的RpcEndpoint都需要注册到RpcEnv实例对象中（注册的时候会指定注册的名称，这样客户端就可以通过名称查询到RpcEndpoint的RpcEndpointRef引用，进而进行通信），在RpcEndpoint接收到消息后会调用receive方法进行处理；

  2. RpcEndpoint如果接受到需要reply的消息的话就会交给自己的receiveAndReply来处理（回复时候是通过RpcCallContext中的reply方法来回复发送者的），如果不需要reply的话就交给receive方法来处理；

  3. RpcEnvFactory是负责创建RpcEnv的，通过RpcEnv.create方法创建RpcEnv实例对象，默认使用的是Netty：

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

  4. RpcEndpoint的生命周期： 构造（constructor）->启动（onStart）、消息接收（receive*）、停止（onStop）

##主要讲解的Class

  • RpcAddress
  • RpcEnvConfig
  • RpcTimeout
  • RpcEndpoint
  • RpcEnvFactory
  • RpcCallContext
  • ThreadSafeRpcEndpoint
  • RpcEndpointRef

自己去看NettyRpcEndpointRef、AkkaRpcEndpointRef