自定义RPC的完整实现---深入理解rpc内部原理

倘若不使用RPC远端调用的情况下，代码如下：

local.py

# coding:utf-8

# 本地调用除法运算的形式
class InvalidOperation(Exception):
    def __init__(self, message = None):
        self.message = message or 'involid operation'

def divide(num1, num2 = 1):
    if num2 == 0:
        raise InvalidOperation
    res = num1 / num2
    return res

try:
    val = divide(200, 100)
except InvalidOperation as e:
    print(e.message)
else:
    print(val)

接下来将使用RPC二进制的形式，远程过程调用代码如下。

service.py 中自定义需要实现消息协议、传输控制，并且实现客户端存根clientStub和服务器端存根serverStub，服务器定义以及channel的定义。

import struct
from io import BytesIO
import socket

class InvalidOperation(BaseException):
    def __init__(self, message = None):
        self.message = message or 'involid operation'

class MethodProtocol(object):
    ''''
    解读方法名
    '''
    def __init__(self, connection):
        self.conn = connection

    def _read_all(self, size):
        """
        帮助我们读取二进制数据
        :param  size: 想要读取的二进制数据大小
        :return:  二进制数据bytes
        """
        # self.conn
        if isinstance(self.conn, BytesIO):
            buff = self.conn.read(size)
            return buff
        else:
            # 有时候长度大于每次读取的长度
            have = 0
            buff = b''
            while have < size:
                chunk = self.conn.recv(size - have)
                buff += chunk
                l = len(chunk)
                have += l
                if l == 0:
                    # 表示客户端已经关闭了
                    raise EOFError
            return buff

    def get_method_name(self):
        # 读取字符串长度
        buff = self._read_all(4)
        length = struct.unpack('!I',buff)[0]

        # 读取字符串
        buff = self._read_all(length)
        name = buff.decode()
        return name

class DivideProtocol(object):
    """
    divide过程消息协议转换工具
    """
    def args_encode(self, num1, num2=1):
        """
        将原始调用的请求参数转换打包成二进制消息数据
        :param num1: int
        :param num2: int
        :return: bytes 二进制消息数据
        """
        name = 'divide'

        # 处理函数名
        buff = struct.pack('!I', 6) # 无符号int
        buff += name.encode()

        # 处理参数1
        buff2 = struct.pack('!B', 1) # 无符号byte
        buff2 += struct.pack('!i', num1)

        # 处理参数2
        if num2 != 1:
            # 没有传参的时候
            buff2 += struct.pack('!B', 2)
            buff2 += struct.pack('!i', num2)

        # 处理参数边界和组合成完整数据
        buff += struct.pack('!I',len(buff2))
        buff += buff2

        return buff

    def _read_all(self, size):
        """
        帮助我们读取二进制数据
        :param  size: 想要读取的二进制数据大小
        :return:  二进制数据bytes
        """
        # self.conn
        if isinstance(self.conn, BytesIO):
            buff = self.conn.read(size)
            return buff
        else:
            # 有时候长度大于每次读取的长度
            have = 0
            buff = b''
            while have < size:
                chunk = self.conn.recv(size - have)
                buff += chunk
                l = len(chunk)
                have +=  l
                if l == 0:
                    # 表示客户端已经关闭了
                    raise EOFError
            return buff

    def args_decode(self, connection):
        """
        接受调用请求数据病进行解析
        :param connection: 链接请求数据 socket  BytesIO
        :return: 因为有多个参数,定义为字典
        """
        param_len_map = {
            1:4,
            2:4,
        }

        param_fmt_map = {
            1:'!i',
            2:'!i',
        }

        param_name_map = {
            1: 'num1',
            2: 'num2',
        }

        # 保存用来返回的参数字典
        args = {}

        self.conn = connection
        # 处理方法的名字,已经提前被处理，稍后处理

        # 处理消息边界
        # 1) 读取二进制数据----read  ， ------ByteIO.read
        # 2) 将二进制数据转换为python的数据类型
        buff = self._read_all(4)
        length = struct.unpack('!I',buff)[0]

        # 记录已经读取的长度值
        have = 0

        # 处理第一个参数
        # 解析参数序号
        buff = self._read_all(1)
        have += 1
        param_seq = struct.unpack('!B', buff)[0]

        # 解析参数值
        param_len = param_len_map[param_seq]
        buff = self._read_all(param_len)
        have += param_len
        param_fmt = param_fmt_map[param_seq]
        param = struct.unpack(param_fmt,buff)[0]

        # 设置解析后的字典
        param_name = param_name_map[param_seq]
        args[param_name] = param

        if have >= length:
            return args
        # 处理第二个参数
        # 解析参数序号
        buff = self._read_all(1)
        param_seq = struct.unpack('!B', buff)[0]

        # 解析参数值
        param_len = param_len_map[param_seq]
        buff = self._read_all(param_len)
        param_fmt = param_fmt_map[param_seq]
        param = struct.unpack(param_fmt, buff)[0]

        # 设置解析后的字典
        param_name = param_name_map[param_seq]
        args[param_name] = param
        return args

    def result_encode(self, result):
        """
        将原始结果数据转换为消息协议二进制数据
        :param result:
        :return:
        """
        if  isinstance(result,float):
            # 处理返回值类型
            buff = struct.pack('!B', 1)
            buff += struct.pack('!f', result)
            return buff
        else:
            buff = struct.pack('!B', 2)
            # 处理返回值
            length = len(result.message)
            # 处理字符串长度
            buff += struct.pack('!I', length)
            buff += result.message.encode()
            return buff

    def result_decode(self, connection):
        """
        将返回值消息数据转换为原始返回值
        :param connection: socket BytesIo
        :return: float InvalidOperation对象
        """
        self.conn = connection
        # 处理返回值类型
        buff = self._read_all(1)
        result_type = struct.unpack('!B', buff)[0]

        if result_type == 1:
            #正常情况
            buff = self._read_all(4)
            val = struct.unpack('!f', buff)[0]
            return val
        else:
            buff = self._read_all(4)
            length = struct.unpack('!I', buff)[0]
            # 读取字符串
            buff = self._read_all(length)
            message = buff.decode(buff)
            return InvalidOperation(message)

class Channel(object):
    """
    用于客户端建立网络链接
    """
    def __init__(self, host, port):
        self.host = host
        self.port = port

    def get_connection(self):
        """
        获取链接对象
        :return: 与服务器通讯的socket
        """
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((self.host, self.port))
        return sock

class Server(object):
    """
    RPC服务器
    """
    def __init__(self, host, port, handlers):
        sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

        # 地址复用
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.host = host
        self.port = port
        # 绑定地址
        sock.bind((self.host, self.port))

        #  因为在启动的方法中才开启监听，所以不在此处开启
        # sock.listen(128)
        self.sock = sock
        self.handlers = handlers

    def serve(self):
        """
        开启服务器运行，提供RPC服务
        :return:
        """
        # 开启服务器的监听，等待客户端的链接请求
        self.sock.listen(128)
        print("服务器开启监听,ip地址为%s,port为%d..." % (self.host,self.port))
        while True:
            # 不断的接收客户端的链接请求
            client_sock, client_addr = self.sock.accept()
            print("与客户端%s建立连接" % str(client_addr))

            # 交个ServerStub,完成客户端的具体的RPC的调用请求
            stub = ServerStub(client_sock, self.handlers)
            try:
                while True:
                    # 不断的接收
                    stub.process()
            except EOFError:
                # 表示客户端关闭了连接
                print('客户端关闭了连接')
                client_sock.close()

class ClientStub(object):
    """
    用来帮助客户端完成远程过程调用 RPC调用

    stub = ClientStub()
    stub.divide(200, 100)
    """
    def __init__(self, channel):
        self.channel = channel
        self.conn = self.channel.get_connection()

    def divide(self, num1, num2 = 1):
        # 将调用的参数打包成消息协议的数据
        proto = DivideProtocol()
        args = proto.args_encode(num1, num2)
        # 将消息数据通过网络发送给服务器
        self.conn.sendall(args)

        # 接受服务器返回的消息数据，并进行解析
        result = proto.result_decode(self.conn)

        # 将结果之（正常float 或 异常InvalidOperation）返回给客户端
        if isinstance(result,float):
            return result
        else:
            raise result

class ServerStub(object):
    """
    服务端存根
    帮助服务端完成远端过程调用
    """
    def __init__(self, connection, handlers):
        """
        :param connection: 与客户端的链接
        :param handlers: 真正的本地函数路由
        此处不以map的形式处理，实现类的形式
        class Handler：
            @staticmethod
            def divide():
                pass
            @staticmethod
            def add():
                pass
        """
        self.conn = connection
        self.method_proto = MethodProtocol(self.conn)
        self.process_map = {
            'divide': self._process_divide,
            'add': self._process_add
        }
        self.handlers = handlers

    def process(self):
        """
        当服务端接受了客户的链接，建立好链接后，完成远端调用的处理
        :return:
        """
        # 接收消息数据，并解析方法的名字
        name = self.method_proto.get_method_name()
        # 根据解析获得的方法名，调用相应的过程协议，接收并解析消息数据
        self.process_map[name]()

    def _process_divide(self):
        """
        处理除法过程调用
        :return:
        """
        proto = DivideProtocol()
        args = proto.args_decode(self.conn)
        # args = {'num1':xxx, 'num2':xxx}
        # 除法过程的本地调用------------------->>>>>>>>>
        # 将本地调用过程的返回值(包括可能的异常)打包成消息协议的数据,通过网络返回给客户端
        try:
            val = self.handlers.divide(**args)
        except InvalidOperation as e:
            ret_message = proto.result_encode(e)
        else:
            ret_message = proto.result_encode(val)
        self.conn.sendall(ret_message)

    def _process_add(self):
        """
        处理加法过程调用
        此方法暂时不识闲
        :return:
        """
        pass

if __name__ == '__main__':
    # 目的：消息协议测试，模拟网络传输
    # 构造消息数据
    proto = DivideProtocol()

    # 测试一
    # divide(200,100)
    # message = proto.args_encode(200,100)

    # 测试二
    message = proto.args_encode(200)

    conn = BytesIO()
    conn.write(message)
    conn.seek(0)

    # 解析消息数据
    method_proto = MethodProtocal(conn)
    name = method_proto.get_method_name()
    print(name)

    args = proto.args_decode(conn)
    print(args)

接下来，只需要创建服务器实例和使用客户端发起请求

server.py

from services import InvalidOperation
from services import Server

class Handlers:
    @staticmethod
    def divide(num1, num2 = 1):
        if num2 == 0:
            raise InvalidOperation('ck_god_err')
        val = num1/num2
        return val

if __name__ == '__main__':
    # 开启服务器
    _server = Server('127.0.0.1', 8000, Handlers)
    _server.serve()

client.py

ffrom services import ClientStub
from services import Channel
from services import InvalidOperation

# 创建与服务器的连接
channel = Channel('127.0.0.1', 8000)

# 创建用于rpc调用的工具
stub = ClientStub(channel)

# 进行调用
for i in range(5):
    try:
        # val = stub.divide(i * 100,100)
        # val = stub.divide(i * 100)
        val = stub.divide( 100, 0)
    except InvalidOperation as e:
        print(e.message)
    else:
        print(val)

当然如果有必要的话，可以在services.py添加如下代码，改为多线程的方式，自己再重写创建实例就可以调用了。

class ThreadServer(object):
    """
    多线成RPC服务器
    """
    def __init__(self, host, port, handlers):
        sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

        # 地址复用
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.host = host
        self.port = port
        # 绑定地址
        sock.bind((self.host, self.port))

        #  因为在启动的方法中才开启监听，所以不在此处开启
        # sock.listen(128)
        self.sock = sock
        self.handlers = handlers

    def serve(self):
        """
        开启服务器运行，提供RPC服务
        :return:
        """
        # 开启服务器的监听，等待客户端的链接请求
        self.sock.listen(128)
        print("服务器开启监听,ip地址为%s,port为%d..." % (self.host,self.port))
        while True:
            # 不断的接收客户端的链接请求
            client_sock, client_addr = self.sock.accept()
            print("与客户端%s建立连接" % str(client_addr))
            t = threading.Thread(target= self.handle, args=(client_sock,))
            t.start()

    # 子线程函数
    def handle(self,client_sock):
        """
        子线程调用的方法，用来处理一个客户段的请求
        :return: 
        """
        # 交个ServerStub,完成客户端的具体的RPC的调用请求
        stub = ServerStub(client_sock, self.handlers)
        try:
            while True:
                # 不断的接收
                stub.process()
        except EOFError:
            # 表示客户端关闭了连接
            print('客户端关闭了连接')
            client_sock.close()