RPC（Remote Procedure Call）即是远程过程调用，简单的理解就是调用远程计算机上的服务，就像调用本地服务一样。而不需要了解底层网络技术的协议。RPC采用的是C/S模式；请求部分是一个客户端，而远程服务提供程序就是一个服务器；客户端首先发送一个函数的请求调用到服务器，然后等待服务器上执行并返回响应信息。在服务器端，首先会保持监听状态直到有客户端请求到达服务端为止，之后服务便会执行调用请求的函数，计算结果并发送响应信息给客户端。

RPC内部结构一般如图所示：

讯享网

1. 客户端程序以本地调用的形式调用本地产生的Stub程序；
2. 该Stub程序会将该函数的调用按照网络通信封装成对应的网络消息包(主要包含调用的类、函数、参数等)发送到远程的服务端；
3. 远程服务端接受到此消息后，会解析还原此消息，并调用对应的服务端的Stub程序；
4. Stub程序会按照被调用的形式调用具体对应的函数以及传递的参数，并将结果返回给Stub程序；
5. 服务端的Stub程序会将此结果封装成消息，通过网络通信的方式返回给客户端程序；

Hadoop RPC

        Hadoop RPC主要由三个大类组成，其即是RPC、Client类和Server类；其主要结构在hadoop-common包下的ipc.RPC类；其类的基本组织结构与提供的函数如下：

Hadoop RPC对外主要提供了两类接口，分别是：

• public static <T>ProtocolProxy <T> getProxy/waitForProxy() : 构造一个客户端代理对象，用于向服务器发送RPC请求
• public static Server RPC.Builder (Configuration).build() : 为某个协议实例构造一个服务器对象，用于处理客户端发送的请求

        其中大量的函数waitForProxy()、getProxy()方法是用来获取对应RPC协议的客户端代理；在waitForProxy()函数中，其会采用java中的动态代理模式。首先会根据选择的序列化协议来初始化一个RpcEngine，主要用于序列化封装函数以及传递的参数；之后便会创建一个叫invoker的RpcInvocationHandler，里面包含了本次连接的remoteId、客户端client、调用协议protocolName等，并且重写了对应的invoke()方法，在invoke()方法中主要是将调用的method方法、调用协议protocolName以及客户端协议版本clientProtocolVersion等封装在rpcRequestHeader中，最终会将对应封装的请求头rpcRequestHeader以及对应的传递args封装成RpcMessageWithHeader；并通过client.call()发送到服务端。在客户端创建好了invocationHandler后，会通过Proxy.newProxyInstance()创建对应的代理类实例，最后在客户端根据生成的代理类实例，即可调用对应的方法。

1、首先在waitForProxy()中其会先根据conf配置文件来使用相应的Rpc引擎来选择不同的序列化方式，其主要是使用RpcKind来标识当前RPC框架使用的引擎，其主要包含WritableRpcEngine和ProtobufRpcEngine；其获取方法getProtocolEngine(protocol, conf).getProxy()如下；我们主要分析ProtobufRpcEngine的方式：

可以看到选择ProtobufRpcEngine序列化的方式，其会构造对应的代理类ProtocolProxy来代理实际的调用者Invoker类；在Invoker类中封装了使用对应序列化方式来序列化客户端请求参数，并通过客户端client，通过socket网络通信来进行序列化参数的发送至远程的服务响应端；Invoker类内部的重要部分如下：

2、在RPC.Builder类中，其是RPC Server的一个构造者对象，可以通过RPC.Builder.build()方法快速构建一个RPC服务端对象。其基本的构建服务端代码如下：

接下来分别介绍一下RPC.Client和RPC.Server端的请求发送、与请求响应的详细过程；

 

RPC Client

       Client类的主要功能就是发送请求和接受响应信息；该Client类只有一个方法入口call()方法。通过waitForProxy()构造的客户端请求代理会调用Client.call()方法将RPC请求发送到远程的服务器上，然后等待远程服务器的响应信息。Client.call()方法发送请求和接受响应的流程如下：

• Client.call()方法用于将RPC请求封装成一个Call对象，Call对象中保存了RPC调用的信息；并且在Call方法中会创建一个用于管理Client和Server端之间的Socket连接的Connection对象；
• 客户端会使用Hashtable<ConnectionId, Connection> connections；来管理和缓存复用对应的socket连接；(与Server端建立Socket连接，会比较消耗资源)；
• Client.call()会调用Connection.setupIOstreams()方法建立Client与Server之间的socket连接，并且会启动Connection线程，其会监听socket并读取server端发回的响应信息；
• Client.call()方法最终会调用Connection.sendRpcRequest()方法来进行RPC请求的发送；
• Client.call()方法会调用Call.wait()方法在Call对象上进行等待，等待Server端返回响应信息；
• Connection线程收到Server端的响应信息后，会设置对应Call对象的返回值字段，并调用Call.notify()唤醒Client.call()方法所在的调用线程读取Call对象的返回值；

Client.call()方法具体的执行源码如下：

 

RPC Server

        Server类即RPC的服务端。Hadoop Server为了保证高性能采用了很多提高并发处理能力的技术，主要包括线程池、事件驱动和Reactor设计模式等；

典型的Reactor设计模式中主要包括以下几个角色 ： 

• Reactor：I/O事件的派发者
• Acceptor：接受来自Client的连接，建立与Client对应的Handler，并向Reactor注册此Handler
• Handler：与一个Client通信的实体，并按一定的过程实现业务的处理
• Reader/Sender：为了加速处理速度，Reactor模式往往构建一个存放数据处理线程的线程池，这样数据读出后，立即扔到线程吃中等待后续处理即可。为此，Reactor模式一般分离Handler中的读和写两个过程，分别注册成单独的读事件和写事件，并由对应的Reader和Sender线程处理

在Hadoop RPC Server的类结构设计中，其也是一个典型的Reactor设计模式：

• Listener：整个Server只有一个Listener线程，用来监听来自客户端的socket请求，Listener对象中定义了一个Selector对象，其负责监听SelectionKey.OP_ACCEPT事件。等待客户端Client.call()中的getConnection建立socket连接来触发该事件唤醒Listener线程，Listener线程会调用ServerSocketChannel.accept()创建一个新的SocketChannel；
• Listener会轮询的从readers线程池中选取一个线程，并在Reader的readerSelector上注册OP_READ事件；
• 当客户端Client发送RPC请求时，其会触发Reader的readerSelector并唤醒Reader线程；
• Reader线程从SocketChannel中读取数据并封装成Call对象，然后放入共享队列callQueue中；
• handlers线程池中的handler线程都通过BlockingQueue.take()阻塞队列方法在callQueue上阻塞，当有Call对象被放入callQueue中后，其中一个Handler线程被唤醒。然后根据Call对象上的信息，调用Server.call()方法，随后会尝试将响应信息写入SocketChannel；
• 当响应结果无法完全写入SocketChannel时，将会在Responder线程的respondSelector上注册OP_WRITE事件，当监听到可写时，会唤醒Responder继续写响应；

接下来一步步来看其RPC.builder.build()；初始化Listener(listener构造初始化时构造Reader线程池)线程、Responder线程等；server.start()；启动rpc server端对应的listener、responder、handlers线程池的详细过程：

Listener类：首先Listener类构造初始化时，会建立socket并绑定监听相关的地址端口后，创建内部的Readers线程池组，并在当前listener上打开Selector，并在channel上注册对SelectionKey.OP_ACCEPT的监听事件。当Server创建Listener并调用start方法启动listener线程后，Listener线程会执行run方法，并循环监听通道上的OP_ACCEPT事件来判断是否有新的连接请求，如果有则调用doAccept()方法来处理；

其中doAccept()方法会接受来自客户端的socket连接请求并初始化socket连接。之后doAccept()会从readers线程池组中轮询的选择出一个Reader线程来读取这个客户端的rpc请求。通过reader.addConnection©将这个Connection对象添加到Reader对象所维护的一个待连接处理队列pendingConnections中，并通过readSelector.wakeup()；唤醒阻塞在reader上的readSelector.select()。此后，这个channel上的读与写任务将一直固定由这个分派给自己的Reader直接负责，而不会被其它Reader线程处理。

Reader类：reader类有自己的readSelector对象。Reader线程的主循环是在执行doRunLoop()方法；当reader.addConnection©有连接添加到reader上并唤醒readSelector.select()后；其会在这个reader的readSelector对象上注册监听SelectionKey.OP_READ事件；之后便会在readSelector上等待可读事件，也就是等待客户端的rpc请求到达，之后便读取该请求并用一个call对象进行封装，最后放入在callQueue中等待handler线程处理；

doRead()函数调用栈如下：

• doRead()->connection.readAndProcess()->processOneRpc()->processRpcRequest()方法

doRead()函数会使用connection对象来进行读取处理，connection对象维护了Server和Client之间的socket连接。reader线程会调用readAndProcess()方法从IO流中读取一个RPC请求，首先其会先从socket流中读取连接头connectionhead，然后读取一个完整的RPC请求，最后会调用processOneRpc()来处理该请求。processOneRpc()会读取出rpc请求头域，然后调用processRpcRequest()处理rpc请求体。

Handler类：是实际处理请求的线程类，负责执行RPC请求对应的本地函数，然后将结果发回客户端。在Server类中会同时存在多个Handler线程，它们并行的从共享队列callqueue中取出待处理的Call对象，然后调用Server.call()方法执行RPC调用对应的本地函数；之后handler会调用setupReponse()方法构造RPC应答结果，并通过reponse.doRespond()将响应结果返回给客户端。

Responder类：也是一个线程类，server端仅有一个Responder对象，其内部包含一个writeSelector对象用于监听channel中的SelectionKey.OP_WRITE事件。此时当Responder线程循环执行doRunLoop()时，其会阻塞等待在writeSelector.select(PURGE_INTERVAL)上，等待通道可写之后，便会相应的执行doRunLoop()->doAsyncWrite()->processResponse()方法来继续执行剩余的响应写操作。

从源码中可以看到，当Handler将call对象加入到当前Connection的responseQueue中时，会判断是否只有当前一个call对象需要返回响应，则将直接在handler线程中调用processResponse()，如果没能将结果一次性返回给客户端时，会在对应的writeSelector上注册SelectionKey.OP_WRITE事件，从而能够使Responder线程采用异步的方式来继续发送未发送完成的结果。