Spark消息通信原理——Spark消息通信架构

Posted 2022-07-04 sysocjs

        在Spark中定义了通信框架的接口，这些接口中调用了Netty的具体方法（在spark2.x前，使用的是Akka）。各接口和实现类的关系如下图所示。

将终端（EndPoint）注册到Rpc环境中：

        在各个模块中，如DriverEndPoint、ClientEndPoint、Master、Worker等，会先使用RpcEnv的静态方法创建RpcEnv实例，然后实例化终端，由于终端都是继承与ThreadSafeEpcEndPoint，即创建的终端实例属于线程安全的，接着调用RpcEnv的启动终端方法setupEndPoint，将终端和其应用的引用注册到RpcEnv中。换句话说，其他对象只要获取终端引用，就可以与其进行通信。

以master.scala为例，startRpcEnvAndEndPoint方法中，启动消息通信框架的源码：

def startRpcEnvAndEndPoint(host:String, port:Int, webUiPort:Int, conf:SparkConf):(RpcEnv, Int, Option[Int]) = 
    val securityMgr = new SecurityManager(conf)
    val rpcEnv = RpcEnv.create(SYSTEM_NAME, host, port, conf, securityMgr)
    val masterEndPoint = rpcEnv.setupEndPoint(ENDPOINT_NAME, new Master(rpcEnv, rpcEnv.address, webUiPort, securityMgr, conf))    //注册master终端
    val portsResponse = masterEndPoint.askWithRetry[BoundPortsResponse](BoundPortsRequest)(rpcEnv, portsResponse.webUIPort, portsResponse.restPort)

        当然，是先有master终端，才会有worker终端。