RPC 学习

RPC 简单介绍

RPC（Remote Procedure Call） 是远程过程调用协议。RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。它的主要目标是让构建分布式计算（应用）更加容易。

RPC 采用客户机/服务器模式。

请求程序就是一个客户机，而服务提供程序就是一个服务器。

它的过程是这样的：

1、客户机调用进程发送一个有进程参数的调用信息给服务进程。
2、调用信息到达服务机后，服务机计算结果，发送答复信息。
3、客户端调用进程接收答复信息，获取进程结果。

我们应该有一些认知：

1、如果我们开发简单的应用，用户不多、流量也不大，就用不着 RPC。
2、如果系统访问量变大、业务增多，可以将业务拆分为几个互不关联的应用，分别部署在不同的机器上，我们也用不着 RPC。
3、如果业务越来越多，发现有些功能无法单独划分，于是就将公共逻辑抽取出来，组成独立的 Service 应用。这时，RPC 就能够让 New Service 与 Public Service 进行程序间的高效通信。

RPC 框架原理

一个 RPC 的程序应该有这几个部分：User、User-stub、RPCRuntime、Server-stub、Server。

粗粒度的过程是：

1、客户端（User）通过本地调用服务。
2、客户端存根（User Stub）接收请求负责将方法、入参等信息序列化（组装）成能够进行网络传输的消息体。
3、客户端 RPCRuntime 实例找到远程服务地址，并将消息通过网络发送给服务端。
4、服务端存根（Server Stub）收到信息后进行反序列化操作（解码）。
5、服务器（Server）根据消息调用服务进行处理，并将处理后的消息返回。

接着，我们再详细看一看 RPC 依赖哪些组件。

RPC 服务通过 RpcServer 暴露（Export）远程接口方法。

RPC 客户通过 RpcClient 引入（Import）远程接口方法。

其中，RPC 框架提供接口的代理实现，实际调用委托给代理 RpcProxy，将封装好的信息交给 RpcInvoker 去实际执行。客户端的 RpcConnector 维持与服务端的通道，并用 RpcProtocal 编码后的信息发送给服务端。

PRC 服务端通过 RpcAcceptor 接收客户端的请求，用 RpcProtocal 进行解码，然后传给 RpcProcessor 去控制处理调用过程，最后委托 RpcInvoker 执行并返回调用结果。

1. RpcServer  
   负责导出（export）远程接口  

2. RpcClient  
   负责导入（import）远程接口的代理实现  

3. RpcProxy  
   远程接口的代理实现  

4. RpcInvoker  
   客户方实现：负责编码调用信息和发送调用请求到服务方并等待调用结果返回  
   服务方实现：负责调用服务端接口的具体实现并返回调用结果  

5. RpcProtocol  
   负责协议编/解码  

6. RpcConnector  
   负责维持客户方和服务方的连接通道和发送数据到服务方  

7. RpcAcceptor  
   负责接收客户方请求并返回请求结果  

8. RpcProcessor  
   负责在服务方控制调用过程，包括管理调用线程池、超时时间等  

9. RpcChannel  
   数据传输通道

RPC 框架核心

服务暴露：远程提供者需要以某种形式提供服务调用相关的信息，包括但不限于服务接口定义、数据结构或者中间态度服务定义文件。

远程代理对象：服务调用者用的服务实际是远程服务的本地代理。

通信：RPC 框架可基于 HTTP 或 TCP 协议，Web Service 是基于 HTTP 协议的 RPC，跨平台性很好但不如基于 TCP 协议的 RPC。

TCP/HTTP：TCP 为传输层协议，HTTP 为应用层协议，越底层越快。
消息 ID：RPC 实际是请求应答模型，长连接方式 TCP 会更加高效，并且定义了消息 ID，可以更容易复用连接。
IO 方式：支持高并发，因此使用异步 IO，就是 NIO。
多连接：因为连接会有发送和接收的缓冲区，如果单连接缓冲不饱和，那么创建多连接，反而增加连接开销。
心跳：连接是由客户端发起建立并维持的。为了保持连接，有必要在 RPC 框架的协议头标记心跳位。

长连接（Long Connector），也称为持久连接，指一个 TCP 连接上可以连续发送多个数据包。如果没有数据包发送，需要发送链路测试包维持连接。

优点：

减少 TCP 连接的建立和关闭的消耗：多个 HTTP 请求可以复用同一个 TCP 连接。

减少网络阻塞：省去较多 TCP 建立和关闭操作，就减少了网络阻塞的影响。

减少 CPU 及内存使用：因为不需要经常建立和关闭连接。

缺点：

系统复杂性提高：需要考虑连接的管理和维护，例如连接的创建、重用和销毁。

资源占用问题：连接数过多，可能会占用过多的系统资源，影响服务端性能和并发数量。

单连接：

单连接模式下，客户端一次只能连接到一个服务器。

在 RPC 框架中，对外可以暴露异步非阻塞 API。

在 Socket 层面，实现多路复用。

多连接：

多连接模式下，客户端可以同时连接到多个服务器。

在 RPC 框架中，对外暴露阻塞式 API。

同一个 RPC 连接不能同时被多个线程使用。

心跳：

心跳是一种定时操作，其中一方（客户端或服务端）会每隔一段固定时间向另一方发送数据包（心跳包或心跳帧）。

心跳就是确认长时间未通信时，互联双方是否在线。

序列化：RPC 的性能受传输方式和序列化的影响。一方面，使用优秀的序列化框架能够提高性能；另一方面，编码内容的信息越少越好，编码规则越简单越好。

-- 调用编码 --  
1. 接口方法  
   包括接口名、方法名  
2. 方法参数  
   包括参数类型、参数值  
3. 调用属性  
   包括调用属性信息，例如调用附件隐式参数、调用超时时间等  
  
-- 返回编码 --  
1. 返回结果  
   接口方法中定义的返回值  
2. 返回码  
   异常返回码  
3. 返回异常信息  
   调用异常信息

除此之外，还需要一些元信息以方便程序编解码以及扩展。

-- 消息头 --  
magic      : 协议魔数，为解码设计  
header size: 协议头长度，为扩展设计  
version    : 协议版本，为兼容设计  
st         : 消息体序列化类型  
hb         : 心跳消息标记，为长连接传输层心跳设计  
ow         : 单向消息标记，  
rp         : 响应消息标记，不置位默认是请求消息  
status code: 响应消息状态码  
reserved   : 为字节对齐保留  
message id : 消息 id  
body size  : 消息体长度  
  
-- 消息体 --  
采用序列化编码，常见有以下格式  
xml   : 如 webservie soap  
json  : 如 JSON-RPC  
binary: 如 thrift; hession; kryo 等

下面是一个 Python 实现的简单 RPC 调用例子：

# 服务端 Server.py
from SimpleXMLRPCServer import SimpleXMLRPCServer

def fun_add(a, b):
    total = a + b
    return total

if __name__ == '__main__':
    s = SimpleXMLRPCServer(('0.0.0.0', 8080))  # 开启xmlrpcserver
    s.register_function(fun_add)  # 注册函数fun_add
    print "server is online..."
    s.serve_forever()  # 开启循环等待

# 客户端 Client.py
from xmlrpclib import ServerProxy  # 导入xmlrpclib的包

s = ServerProxy("http://172.171.5.205:8080")  # 定义xmlrpc客户端
print s.fun_add(2,3)  # 调用服务器端的函数

RPC 框架实例

我们实现一个简单的 RPC 框架，这个框架包括一个服务器和一个客户端。

服务器使用一个字典存储远程调用的函数，客户端通过网络发送函数名和参数来调用函数。

# 服务器端
import socket
import json
import threading

class Server:
    def __init__(self, host='localhost', port=5000):
        self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server.bind((host, port))
        self.server.listen(5)
        self.functions = {}

    def register_function(self, function):
        self.functions[function.__name__] = function

    def handle_client(self, client):
        while True:
            data = json.loads(client.recv(1024).decode())
            function_name = data['function_name']
            args = data['args']
            result = self.functionsfunction_name
            client.send(json.dumps(result).encode())

    def run(self):
        while True:
            client, _ = self.server.accept()
            threading.Thread(target=self.handle_client, args=(client,)).start()

# 客户端
class Client:
    def __init__(self, host='localhost', port=5000):
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.connect((host, port))

    def call(self, function_name, *args):
        self.client.send(json.dumps({'function_name': function_name, 'args': args}).encode())
        result = json.loads(self.client.recv(1024).decode())
        return result

当然 RPC 还是有很多优化方向的：

长连接/短连接：如果每次调用 RPC 接口都要开启一个 Socket 建立连接损耗就很多了。
服务端线程池：服务端是单线程的时候，每次都要等一个请求处理完才能 accept 下一个 Socket 连接，如果能用线程池进行处理，就可以同时处理多个 RPC 请求。
服务注册中心：调用服务，就需要先注册一个服务中心，告诉对方服务都有哪些实例。
负载均衡：负责选择多个实例中的其中一个进行调用。
结果缓存：每次查询接口时都一定去服务端查询吗，是否能将一些内容进行缓存。
多版本控制：如果服务端接口修改了，旧的接口可以通过版本号来区分。
异步调用：客户端调用完接口后，不想等待服务端返回，就需要支持异步调用。
停机优化：服务端要停机了，但是还没处理完请求，可以先停止接收新请求，将所有请求处理完毕。