快捷導(dǎo)航

通過實(shí)例解析Python RPC實(shí)現(xiàn)原理及方法

更新時(shí)間：2020年07月07日 10:10:22 作者：cxc1357

這篇文章主要介紹了通過實(shí)例解析Python RPC實(shí)現(xiàn)原理及方法,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

單線程同步

使用socket傳輸數(shù)據(jù)
使用json序列化消息體
struct將消息編碼為二進(jìn)制字節(jié)串，進(jìn)行網(wǎng)絡(luò)傳輸

消息協(xié)議

// 輸入
{
  in: "ping",
  params: "ireader 0"
}

// 輸出
{
  out: "pong",
  result: "ireader 0"
}

客戶端 client.py

# coding: utf-8
# client.py

import json
import time
import struct
import socket


def rpc(sock, in_, params):
  response = json.dumps({"in": in_, "params": params}) # 請求消息體
  length_prefix = struct.pack("I", len(response)) # 請求長度前綴
  sock.sendall(length_prefix)
  sock.sendall(response)
  length_prefix = sock.recv(4) # 響應(yīng)長度前綴
  length, = struct.unpack("I", length_prefix)
  body = sock.recv(length) # 響應(yīng)消息體
  response = json.loads(body)
  return response["out"], response["result"] # 返回響應(yīng)類型和結(jié)果

if __name__ == '__main__':
  s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  s.connect(("localhost", 8080))
  for i in range(10): # 連續(xù)發(fā)送10個(gè)rpc請求
    out, result = rpc(s, "ping", "ireader %d" % i)
    print out, result
    time.sleep(1) # 休眠1s，便于觀察
  s.close() # 關(guān)閉連接

服務(wù)端 blocking_single.py

# coding: utf8
# blocking_single.py

import json
import struct
import socket


def handle_conn(conn, addr, handlers):
  print addr, "comes"
  while True: # 循環(huán)讀寫
    length_prefix = conn.recv(4) # 請求長度前綴
    if not length_prefix: # 連接關(guān)閉了
      print addr, "bye"
      conn.close()
      break # 退出循環(huán)，處理下一個(gè)連接
    length, = struct.unpack("I", length_prefix)
    body = conn.recv(length) # 請求消息體
    request = json.loads(body)
    in_ = request['in']
    params = request['params']
    print in_, params
    handler = handlers[in_] # 查找請求處理器
    handler(conn, params) # 處理請求


def loop(sock, handlers):
  while True:
    conn, addr = sock.accept() # 接收連接
    handle_conn(conn, addr, handlers) # 處理連接


def ping(conn, params):
  send_result(conn, "pong", params)


def send_result(conn, out, result):
  response = json.dumps({"out": out, "result": result}) # 響應(yīng)消息體
  length_prefix = struct.pack("I", len(response)) # 響應(yīng)長度前綴
  conn.sendall(length_prefix)
  conn.sendall(response)


if __name__ == '__main__':
  sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 創(chuàng)建一個(gè)TCP套接字
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 打開reuse addr選項(xiàng)
  sock.bind(("localhost", 8080)) # 綁定端口
  sock.listen(1) # 監(jiān)聽客戶端連接
  handlers = { # 注冊請求處理器
    "ping": ping
  }
  loop(sock, handlers) # 進(jìn)入服務(wù)循環(huán)

多線程同步

使用線程庫thread創(chuàng)建原生線程
服務(wù)器可并行處理多個(gè)客戶端

服務(wù)端 multithread.py

多進(jìn)程同步

Python的GIL導(dǎo)致單個(gè)進(jìn)程只能占滿一個(gè)CPU核心，多線程無法利用多核優(yōu)勢
os.fork()會生成子進(jìn)程
子進(jìn)程退出后，父進(jìn)程需使用waitpid系統(tǒng)調(diào)用收割子進(jìn)程，防止其稱為僵尸資源
在子進(jìn)程中關(guān)閉服務(wù)器套接字后，在父進(jìn)程中也要關(guān)閉服務(wù)器套接字
因?yàn)檫M(jìn)程fork后，父子進(jìn)程都有自己的套接字引用指向內(nèi)核的同一份套接字對象，套接字引用計(jì)數(shù)為2，對套接字進(jìn)程close，即將套接字對象的引用計(jì)數(shù)減1

PreForking同步