1.RPC基本原理

RPC

1.界说

远程过程调用(remote procedure call)
2.概念

广义:全部通过网络进行通讯,的调用统称为RPC调用
狭义:不采用http协议的方式,采用自界说格式的二进制方式
3.优缺点

• 优点
  
  
  
  • 服从高
    
  • 发起rpc调用的一方,可以忽略RPC的具体实现,犹如编写当地函数调用
    
  
  
• 缺点
  
  
  
  • 通用性不高
    
  
  

4.RPC结构

• client(caller):调用者
  
• client stub(bundle args/unbundle ret vals):客户端存根
  
• client network service
  
• server network service
  
• server stub(bundle ret vals/unbundle args)
  

5.RPC消息协议

5.1 消息界限

• 分隔符(\r\n)
  
• 长度声明法(例如HTTP中 Content-Length)
  

5.2 内容

• 二进制
  
• 文本内容
  

5.3 压缩

• 压缩处理是一把双刃剑,淘汰数据量减轻带宽压力同时,额外增长了压缩和解压缩的时间
  

6.RPC的实现

6.1 divide_protocol.py

1. import struct
2. from io import BytesIO
5. class InvalidOperation(Exception):
6.     ...
9. class DivideProtocol(object):
10.     """
11.     float divide(1:int num1, 2:int num2=1)
12.     """
14.     def _read_all(self, size):
15.         """
16.         读取指定长度的字节
17.         :param size: 长度
18.         :return: 读取出的二进制数据
19.         """
20.         if isinstance(self.conn, BytesIO):
21.             # BytesIO类型，用于演示
22.             buff = b''
23.             have = 0
24.             while have < size:
25.                 chunk = self.conn.read(size - have)
26.                 have += len(chunk)
27.                 buff += chunk
28.             return buff
30.         else:
31.             # socket类型
32.             buff = b''
33.             have = 0
34.             while have < size:
35.                 chunk = self.conn.recv(size - have)
36.                 have += len(chunk)
37.                 buff += chunk
38.                 # 客户端关闭了连接
39.                 if len(chunk) == 0:
40.                     raise EOFError()
41.             return buff
43.     def args_encode(self, num1, num2=1):
44.         """
45.         对调用参数进行编码
46.         :param num1: int
47.         :param num2: int
48.         :return: 编码后的二进制数据
49.         """
50.         # 处理参数num1, 4字节整型
51.         buff = struct.pack('!B', 1)
52.         buff += struct.pack('!i', num1)
54.         # 处理参数num2, 4字节整型，如为默认值1，则不再放到消息中
55.         if num2 != 1:
56.             buff += struct.pack('!B', 2)
57.             buff += struct.pack('!i', num2)
59.         # 处理消息总长度，4字节无符号整型
60.         length = len(buff)
62.         # 处理方法名，字符串类型
63.         name = 'divide'
64.         # 字符串长度，4字节无符号整型
65.         msg = struct.pack('!I', len(name))
66.         msg += name.encode()
68.         msg += struct.pack('!I', length) + buff
70.         return msg
72.     def args_decode(self, connection):
73.         """
74.         获取调用参数并进行解码
75.         :param connection: 传输工具对象，如socket对象或者BytesIO对象，从中可以读取消息数据
76.         :return: 解码后的参数字典
77.         """
78.         # 保存到当前对象中，供_read_all方式使用
79.         self.conn = connection
80.         param_name_map = {
81.             1: 'num1',
82.             2: 'num2'
83.         }
84.         param_len_map = {
85.             1: 4,
86.             2: 4
87.         }
88.         # 用于保存解码后的参数字典
89.         args = dict()
91.         # 读取消息总长度，4字无节符号整数
92.         buff = self._read_all(4)
93.         length = struct.unpack('!I', buff)[0]
95.         # 记录已读取的长度
96.         have = 0
98.         # 读取第一个参数，4字节整型
99.         buff = self._read_all(1)
100.         have += 1
101.         param_seq = struct.unpack('!B', buff)[0]
102.         param_len = param_len_map[param_seq]
103.         buff = self._read_all(param_len)
104.         have += param_len
105.         args[param_name_map[param_seq]] = struct.unpack('!i', buff)[0]
107.         if have >= length:
108.             return args
110.         # 读取第二个参数，4字节整型
111.         buff = self._read_all(1)
112.         have += 1
113.         param_seq = struct.unpack('!B', buff)[0]
114.         param_len = param_len_map[param_seq]
115.         buff = self._read_all(param_len)
116.         have += param_len
117.         args[param_name_map[param_seq]] = struct.unpack('!i', buff)[0]
119.         return args
121.     def result_encode(self, result):
122.         """
123.         对调用的结果进行编码
124.         :param result: float 或 InvalidOperation对象
125.         :return: 编码后的二进制数据
126.         """
127.         if isinstance(result, float):
128.             # 没有异常，正常执行
129.             # 处理结果类型，1字节无符号整数
130.             buff = struct.pack('!B', 1)
132.             # 处理结果值, 4字节float
133.             buff += struct.pack('!f', result)
134.         else:
135.             # 发生了InvalidOperation异常
136.             # 处理结果类型，1字节无符号整数
137.             buff = struct.pack('!B', 2)
139.             # 处理异常结果值, 字符串
140.             # 处理字符串长度, 4字节无符号整数
141.             buff += struct.pack('!I', len(result.message))
142.             # 处理字符串内容
143.             buff += result.message.encode()
145.         return buff
147.     def result_decode(self, connection):
148.         """
149.         对调用结果进行解码
150.         :param connection: 传输工具对象，如socket对象或者BytesIO对象，从中可以读取消息数据
151.         :return: 结果数据
152.         """
153.         self.conn = connection
155.         # 取出结果类型, 1字节无符号整数
156.         buff = self._read_all(1)
157.         result_type = struct.unpack('!B', buff)[0]
158.         if result_type == 1:
159.             # float的结果值， 4字节float
160.             buff = self._read_all(4)
161.             result = struct.unpack('!f', buff)[0]
162.             return result
163.         else:
164.             # InvalidOperation对象
165.             # 取出字符串长度, 4字节无符号整数
166.             buff = self._read_all(4)
167.             str_len = struct.unpack('!I', buff)[0]
168.             buff = self._read_all(str_len)
169.             message = buff.decode()
170.             return InvalidOperation(message)
173. class MethodProtocol(object):
174.     def __init__(self, connection):
175.         self.conn = connection
177.     def _read_all(self, size):
178.         """
179.         读取指定长度的字节
180.         :param size: 长度
181.         :return: 读取出的二进制数据
182.         """
183.         if isinstance(self.conn, BytesIO):
184.             # BytesIO类型，用于演示
185.             buff = b''
186.             have = 0
187.             while have < size:
188.                 chunk = self.conn.read(size - have)
189.                 have += len(chunk)
190.                 buff += chunk
192.             return buff
194.         else:
195.             # socket类型
196.             buff = b''
197.             have = 0
198.             while have < size:
199.                 print('have=%d size=%d' % (have, size))
200.                 chunk = self.conn.recv(size - have)
201.                 have += len(chunk)
202.                 buff += chunk
204.                 if len(chunk) == 0:
205.                     raise EOFError()
207.             return buff
209.     def get_method_name(self):
210.         # 获取方法名
211.         # 读取字符串长度，4字节无符号整型
212.         buff = self._read_all(4)
213.         str_len = struct.unpack('!I', buff)[0]
215.         # 读取字符串
216.         buff = self._read_all(str_len)
217.         name = buff.decode()
218.         return name

复制代码

6.2 server.py

1. import socket
2. import threading
4. from customize_rpc.divide_protocol import DivideProtocol, MethodProtocol, InvalidOperation
7. class Handlers:
8.     @staticmethod
9.     def divide(num1, num2=1):
10.         """
11.         除法
12.         :param num1:
13.         :param num2:
14.         :return:
15.         """
16.         if num2 == 0:
17.             raise InvalidOperation()
18.         val = num1 / num2
19.         return val
22. class ServerStub(object):
23.     def __init__(self, connection, handlers):
24.         """
25.         服务器存根
26.         :param connection: 与客户端的socket连接
27.         :param handlers: 存放被调用的方法
28.         """
29.         self._process_map = {
30.             'divide': self._process_divide,
31.         }
32.         self.conn = connection
33.         self.method_proto = MethodProtocol(self.conn)
34.         self.handlers = handlers
36.     def process(self):
37.         """
38.         被服务器调用的入口，服务器收到请求后调用该方法
39.         """
40.         # 获取解析调用请求的方法名
41.         name = self.method_proto.get_method_name()
43.         # 调用对应的处理方法
44.         self._process_map[name]()
46.     def _process_divide(self):
47.         """
48.         执行divide本地调用，并将结果返回给客户端
49.         """
50.         # 接收调用参数
51.         proto = DivideProtocol()
52.         args = proto.args_decode(self.conn)
54.         # 进行本地divide调用
55.         try:
56.             result = self.handlers.divide(**args)
57.         except InvalidOperation as e:
58.             result = e
60.         # 构造返回值消息并返回
61.         result = proto.result_encode(result)
62.         self.conn.sendall(result)
65. class Server(object):
66.     def __init__(self, host, port, handlers):
67.         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
68.         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
69.         self.host = host
70.         self.port = port
71.         self.sock.bind((host, port))
72.         self.handlers = handlers
74.     def serve(self):
75.         """
76.         开始服务
77.         """
78.         self.sock.listen(128)
79.         print("开始监听")
80.         while True:
81.             conn, addr = self.sock.accept()
82.             print("建立链接%s" % str(addr))
83.             stub = ServerStub(conn, self.handlers)
84.             try:
85.                 while True:
86.                     stub.process()
87.             except EOFError:
88.                 print("客户端关闭连接")
89.             # 关闭服务端连接
90.             conn.close()
93. class ThreadServer(object):
94.     def __init__(self, host, port, handlers):
95.         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
96.         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
97.         self.host = host
98.         self.port = port
99.         self.sock.bind((host, port))
100.         self.handlers = handlers
102.     def serve(self):
103.         """
104.         开始服务
105.         """
106.         self.sock.listen(128)
107.         print("开始监听")
108.         while True:
109.             conn, addr = self.sock.accept()
110.             print("建立链接%s" % str(addr))
111.             t = threading.Thread(target=self.handle, args=(conn,))
112.             t.start()
114.     def handle(self, client):
115.         stub = ServerStub(client, self.handlers)
116.         try:
117.             while True:
118.                 stub.process()
119.         except EOFError:
120.             print("客户端关闭连接")
122.         client.close()
125. if __name__ == '__main__':
126.     server = Server('127.0.0.1', 8000, Handlers)
127.     server.serve()   

复制代码

6.3 client.py

1. import time
2. import socket
4. from customize_rpc.divide_protocol import DivideProtocol, InvalidOperation
7. class Channel(object):
8.     """
9.     连接通道
10.     """
12.     def __init__(self, host, port):
13.         self.host = host
14.         self.port = port
16.     def get_connection(self):
17.         """
18.         获取一个tcp连接
19.         """
20.         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
21.         sock.connect((self.host, self.port))
22.         return sock
25. class ClientStub(object):
26.     """
27.     客户端存根
28.     """
30.     def __init__(self, channel: Channel):
31.         self.channel = channel
32.         self.conn = self.channel.get_connection()
34.     def divide(self, num1, num2=1):
35.         # 构造
36.         proto = DivideProtocol()
37.         args = proto.args_encode(num1, num2)
38.         self.conn.sendall(args)
39.         result = proto.result_decode(self.conn)
40.         if isinstance(result, InvalidOperation):
41.             raise result
42.         else:
43.             return result
46. if __name__ == '__main__':
47.     channel = Channel('127.0.0.1', 8000)
48.     stub = ClientStub(channel)
50.     for i in range(5):
51.         try:
52.             val = stub.divide(i * 100, 10)
53.         except InvalidOperation as e:
54.             print(e.message)
55.         else:
56.             print(val)
57.         time.sleep(1)

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。