自定义RPC的完整实现---深入理解rpc内部原理
admin
2023-07-18 02:02:24
0

倘若不使用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()

相关内容

热门资讯

矛头指向万斯,以色列谍报界:都... 【文/观察者网 阮佳琪】美国急于从对伊战争中抽身,被特朗普甩脸子的以色列开始各种找补。据《耶路撒冷邮...
一项技术获国家发明专利授权 来源:青海日报 本报格尔木讯 (记者 冉志敏 通讯员 贺昕) 6月5日,从五矿盐湖有限公司传来好消息...
魔都看展丨用无边创意“加载”现... UFO Terminal发起的“加载...计划”步入第三年,油罐艺术公园1号罐正在举办展览“加载…权...
俄总统顾问:美国正试图退出乌克... 据凤凰卫视报道,俄罗斯总统新闻秘书佩斯科夫6月6日表示,就乌克兰问题而言,达成特别军事行动既定目标、...
白龙马申请客户端域名切换方法专... 国家知识产权局信息显示,北京白龙马云行科技有限公司申请一项名为“一种客户端域名切换方法、装置、电子设...
魅族22全面断货,国产智能手机... 作为国产智能手机鼻祖,魅族曾经是多少人心目中的“白月光”?可今年魅族却迎来落幕时刻。魅族今年2月正式...
原创 这... 最近看手机价格,我最大的感受就是:现在买手机真的不能只看谁刚发布。很多新机刚上市时声量很大,价格也很...
经批准,这位考生高考每科可增加... 6月6日下午,极目新闻记者来到巴东县第二高级中学高考考点。因出生时缺氧导致脑瘫的考生小雅(化名),正...
微软官宣智能体硬件:AI时代的... 2026台北国际电脑展(Computex)上,微软接连官宣超20款重磅更新,涉及9款自研模型、与英伟...
汤道生×姚顺雨:腾讯AI下半场... 作者:王聪彬 大会还在进行中,AI已经开始“交作业”了。 在2026腾讯云AI产业应用大会直播期间,...