Spark-RPC理解

基本架构

Akka Actor式RPC架构

1. Spark采用的是AkkaActor架构实现RPC，可是实际使用过程为了兼容不一样节点之间的文件下载，采用Netty来实现Actor功能。
2. Spark RPC由三部分组成：

• RpcEnv RPC的执行上下文，等同于ActorSystem,用于管理RpcEndpoint和RpcEndpointRef
• RpcEndpoint RPC通讯实体的抽象，等同于Actor，用于接收客户端发送来的请求，方法主要有receive,onConnected, onDisconnnected, onStart, onStop, onError等

• RpcEndpointRef RPC通讯实体的引用，等同于ActorRef，在客户端被调用，用来向服务端请求，主要方法是ask和askWithRetry

  核心组件

  Dispatcher

• InboxMessage: 外部发送过来的消息(onStart, onStop, RPCMessage, OneWayMessage...)
• EndpointData: 包装(RpcEndpoint, NettyRpcEndpointRef, Inbox(InboxMessge队列))
• MessageLoop: 经过线程池调度，读取阻塞队列中是否有消息，有的话就直接读取，不然阻塞
• Inbox来源（消息来源）:
• [x] - 注册RpcEndpoint（会生成OnStart消息）
• [x] - 去注册RpcEndpoint(会生成onStop消息)
• [x] - postMessage, 投递消息给指定的RpcEndpoint

• [x] - 中止Dispatcher

  TransportClientFactory

  RPC客户端的工厂类，用于批量生成TransportClient
• ClientPool，ClientFactory内部经过<sokectAddress, ClientPool> 创建套接字(Socket网络链接)与ClientPool(TransportClient)的关联，同时经过object与TransportClient创建1V1的锁关联关系；即对于一个socket，会有多个TransportClient与其关联，spark经过每个TransportClient使用不一样的lock(object)，来进行并行，本质仍是一个利用线程池（链接缓冲池）的思想
  其类型定义为

class TransClientFactory {
    ConcurrentHashMap<SocketAddress, ClientPool> connectionPool;
}
class ClientPool {
    TransportClient[] clients;
    Object[] locks;
}

• TransportClient
  包含5种发送消息的方法: fetchChunk, stream, sendRPC, sendRPCSyns, send

• TransportClientBootstrap 由TransportContext传入，启动加载（如 SAAL和加密认证之类的启动操做）

  TransportContext

  经过createClientFactory建立TransportClientFactory，间接经过createClient建立TransportClient; 经过createServer建立TransServer实例
• TransportConf 配置稳健加载
• RpcHandler，是一个abstract类，实现类为NettyRpcHandler,internalReceive负责将ByteBuffer转换成RequestMessage; postMessage用于投递消息， 而后交由对应的RPCEndpoint处理

val msgDispatch = internalReceive(client, message)
dispatcher.postMessage(msgDispatch, callback)

• NettyStreamMessage: 提供文件服务能力

  NettyRPCEnv

• timeoutScheduler 超时请求的调度器，使用的ScheduleredExcutorService
• clientConnectExecutor
• outboxes: 在send()时在messages中add消息，而后调用drainOutbox()循环遍历发送messages中全部消息;drainOutbox()在没有client时会调用launchuConnectTask()建立TransportClient

private val outboxes = new ConcurrentHashMap[RpcAddress, Outbox]()
class Outbox {
    nettyEnv;   //所在环境
    address;    //远端NettyRpcEnv地址
    messages;   //向外发送的消息列表
    client; // TransportClient
    connectFuture; //链接任务的Future引用
    stopped; //是否中止
    draining;   //Outbox正有线程处理消息
}

• RPC客户端发送请求流程

1. 调用NettyRpcEndpointRef的send/ask方法向RpcEndpoint发送消息;

a) 若是是同一节点，直接使用Dispatcher的postLocalMessage和postOneWayMessage，直接将消息放入EndpointData的Inbox中；

b) 若是发送方在远处，将消息封装成OutboxMessage，放入远端RpcEndpoint对应的Outbox的messages列表中；

1. Outbox的drainOutbox循环从messages获取OutboxMessage，调用TransportClient向远端发送消息;

2. 与远端的TransportServer创建链接以后，经Netty管道，NettyRpcHandler处理，投递到远端的Dispatcher的EndpointData的Inbox中进行处理

   TransportServer

   -TransportRequestHandler：主要是handle()方法，该方法根据request的类型，调用不一样的 processXX()方法进行处理

processFetchRequest 处理获取块请求
processRPCRequest 处理RPC请求
processStreamRequest 处理Stream请求
processOneWayMessage 处理无需回复的请求

RPC服务端实现

• TransportServer

  要点总结

1. Spark RPC是用Netty实现了数据流传输，以及Actor这种RPC框架的，其中NettyRpcEnv至关于ActorySysm, RpcEndpoint至关于Actor（远端的服务，或者说接口，注册在服务端）, RpcEndpointRef至关于ActorRef（服务引用，在客户端使用），双方通讯经过Message这个载体;
2. 客户端发送消息时，经过<address, Outbox[messages, client]>这种结构，向address不断地发送消息；
3. 服务端经过NettyRpcHandler进行消息的receive，转换成InboxMessage，放入Dispatcher中，Dispatcher使用messageLoop循环遍历Inbox，取出InboxMessage，根据消息路由，调用相应方法进行处理，即路由功能