码迷,mamicode.com
首页 > 编程语言 > 详细

[Python] RPC实现

时间:2020-07-05 10:54:57      阅读:86      评论:0      收藏:0      [点我收藏+]

标签:处理   进程同步   bin   ada   display   listen   查找   client   print   

单线程同步

  • 使用socket传输数据
  • 使用json序列化消息体
  • struct将消息编码为二进制字节串,进行网络传输

消息协议

技术图片
 1 // 输入
 2 {
 3     in: "ping",
 4     params: "ireader 0"
 5 }
 6 
 7 // 输出
 8 {
 9     out: "pong",
10     result: "ireader 0"
11 }
View Code

客户端  client.py

技术图片
 1 # coding: utf-8
 2 # client.py
 3 
 4 import json
 5 import time
 6 import struct
 7 import socket
 8 
 9 
10 def rpc(sock, in_, params):
11     response = json.dumps({"in": in_, "params": params})  # 请求消息体
12     length_prefix = struct.pack("I", len(response)) # 请求长度前缀
13     sock.sendall(length_prefix)
14     sock.sendall(response)
15     length_prefix = sock.recv(4)  # 响应长度前缀
16     length, = struct.unpack("I", length_prefix)
17     body = sock.recv(length) # 响应消息体
18     response = json.loads(body)
19     return response["out"], response["result"]  # 返回响应类型和结果
20 
21 if __name__ == __main__:
22     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
23     s.connect(("localhost", 8080))
24     for i in range(10): # 连续发送10个rpc请求
25         out, result = rpc(s, "ping", "ireader %d" % i)
26         print out, result
27         time.sleep(1)  # 休眠1s,便于观察
28     s.close() # 关闭连接
View Code

技术图片

服务端  blocking_single.py

技术图片
 1 # coding: utf8
 2 # blocking_single.py
 3 
 4 import json
 5 import struct
 6 import socket
 7 
 8 
 9 def handle_conn(conn, addr, handlers):
10     print addr, "comes"
11     while True:  # 循环读写
12         length_prefix = conn.recv(4)  # 请求长度前缀
13         if not length_prefix:  # 连接关闭了
14             print addr, "bye"
15             conn.close()
16             break  # 退出循环,处理下一个连接
17         length, = struct.unpack("I", length_prefix)
18         body = conn.recv(length)  # 请求消息体  
19         request = json.loads(body)
20         in_ = request[in]
21         params = request[params]
22         print in_, params
23         handler = handlers[in_]  # 查找请求处理器
24         handler(conn, params)  # 处理请求
25 
26 
27 def loop(sock, handlers):
28     while True:
29         conn, addr = sock.accept()  # 接收连接
30         handle_conn(conn, addr, handlers)  # 处理连接
31 
32 
33 def ping(conn, params):
34     send_result(conn, "pong", params)
35 
36 
37 def send_result(conn, out, result):
38     response = json.dumps({"out": out, "result": result})  # 响应消息体
39     length_prefix = struct.pack("I", len(response))  # 响应长度前缀
40     conn.sendall(length_prefix)
41     conn.sendall(response)
42 
43 
44 if __name__ == __main__:
45     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 创建一个TCP套接字
46     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 打开reuse addr选项
47     sock.bind(("localhost", 8080)) # 绑定端口
48     sock.listen(1)  # 监听客户端连接
49     handlers = {  # 注册请求处理器
50         "ping": ping
51     }
52     loop(sock, handlers)  # 进入服务循环
View Code

技术图片

多线程同步

  • 使用线程库thread创建原生线程
  • 服务器可并行处理多个客户端

服务端  multithread.py

技术图片View Code

技术图片

多进程同步 

  • Python的GIL导致单个进程只能占满一个CPU核心,多线程无法利用多核优势
  • os.fork()会生成子进程
  • 子进程退出后,父进程需使用waitpid系统调用收割子进程,防止其称为僵尸资源
  • 在子进程中关闭服务器套接字后,在父进程中也要关闭服务器套接字
  • 因为进程fork后,父子进程都有自己的套接字引用指向内核的同一份套接字对象,套接字引用计数为2,对套接字进程close,即将套接字对象的引用计数减1

服务端  multiprocess.py

技术图片View Code

PreForking同步

  • 进程比线程耗费资源,通过PreForking进程池模型对服务器开辟的进程数量进行限制,避免服务器负载过重
  • 如果并行的连接数量超过了prefork进程数量,后来的客户端请求将会阻塞

单进程异步

  • 通过事件轮询API,查询相关套接字是否有响应的读写事件,有则携带事件列表返回,没有则阻塞
  • 拿到读写事件后,可对事件相关的套接字进行读写操作
  • 设置读写缓冲区
  • Nginx/Nodejs/Redis都是基于异步模型
  • 异步模型编码成本高,易出错,通常在公司业务代码中采用同步模型,仅在讲究高并发高性能的场合才使用异步模型 

PreForking异步

  •  Tornado/Nginx采用了多进程PreForking异步模型,具有良好的高并发处理能力

技术图片 

参考

Python多线程和多进程

https://www.cnblogs.com/yssjun/p/11302500.html

[Python] RPC实现

标签:处理   进程同步   bin   ada   display   listen   查找   client   print   

原文地址:https://www.cnblogs.com/cxc1357/p/13197183.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!