Python 基于 multiprocessing 实现远程方法调用 (RPC) 详解

基于 multiprocessing 的 RPC 实现原理

在 Python 中实现远程方法调用，最基础且高效的方式是利用 multiprocessing 库提供的连接机制。该方案的核心在于将函数请求、参数及返回值进行序列化（通常使用 pickle），通过网络连接传输，并在服务端反序列化后执行对应函数。整个过程涉及四个关键组件：RPC 处理器、服务器监听、客户端代理以及通信连接。

1. 定义 RPC 处理器 (RPCHandler)

服务端需要一个处理器来管理注册的函数。当客户端发起请求时，处理器负责接收消息、反序列化参数、查找并执行对应的函数，最后将结果或异常序列化后返回。

# rpcserver.py

import pickle
class RPCHandler:
  def __init__(self):
    self._functions = { }

  def register_function(self, func):
    self._functions[func.__name__] = func

  def handle_connection(self, connection):
    try:
      while True:
        # Receive a message
        func_name, args, kwargs = pickle.loads(connection.recv())
        # Run the RPC and send a response
        try:
          r = self._functions[func_name](*args,**kwargs)
          connection.send(pickle.dumps(r))
        except Exception as e:
          connection.send(pickle.dumps(e))
    except EOFError:
       pass

在 handle_connection 方法中，程序持续监听连接。一旦收到数据，便使用 pickle.loads() 还原出函数名、位置参数和关键字参数。执行完成后，结果通过 connection.send() 发送。若执行过程中抛出异常，异常对象也会被序列化并发送回客户端，确保客户端能捕获到具体的错误信息。

2. 启动 RPC 服务器

服务器端利用 multiprocessing.connection.Listener 监听指定地址和端口。为了支持并发处理多个客户端请求，服务器为每个新连接启动一个守护线程，并调用处理器的 handle_connection 方法。

from multiprocessing.connection import Listener
from threading import Thread

def rpc_server(handler, address, authkey):
  sock = Listener(address, authkey=authkey)
  while True:
    client = sock.accept()
    t = Thread(target=handler.handle_connection, args=(client,))
    t.daemon = True
    t.start()

# Some remote functions
def add(x, y):
  return x + y

def sub(x, y):
  return x - y

# Register with a handler
handler = RPCHandler()
handler.register_function(add)
handler.register_function(sub)

# Run the server
rpc_server(handler, ('localhost', 17000), authkey=b'peekaboo')

上述代码注册了 add 和 sub 两个远程函数，并通过 rpc_server 函数在本地端口 17000 上启动服务。注意使用了 authkey 进行身份验证，这是防止未授权访问的第一道防线。

3. 创建 RPC 代理 (RPCProxy)

客户端通过 RPCProxy 类来模拟本地调用。该类的核心在于 __getattr__ 方法的重写。当客户端调用代理对象的任意属性（即远程函数名）时，__getattr__ 会被触发，并返回一个闭包函数 do_rpc。

import pickle

class RPCProxy:
  def __init__(self, connection):
    self._connection = connection
  def __getattr__(self, name):
    def do_rpc(*args, **kwargs):
      self._connection.send(pickle.dumps((name, args, kwargs)))
      result = pickle.loads(self._connection.recv())
      if isinstance(result, Exception):
        raise result
      return result
    return do_rpc

在 do_rpc 闭包中，函数名和参数被打包成元组，经 pickle.dumps() 序列化后通过网络发送。随后，程序阻塞等待服务端的响应。如果返回的是异常对象，则直接抛出；否则返回计算结果。

4. 客户端调用示例

客户端使用 Client 连接到服务器，创建代理对象后即可像调用本地函数一样调用远程方法。

>>> from multiprocessing.connection import Client
>>> c = Client(('localhost', 17000), authkey=b'peekaboo')
>>> proxy = RPCProxy(c)
>>> proxy.add(2, 3)

5
>>> proxy.sub(2, 3)
-1
>>> proxy.sub([1, 2], 4)
Traceback (most recent call last):
 File "<stdin>", line 1, <module>
 File "rpcserver.py", line 37, in do_rpc
  raise result
TypeError: unsupported operand type(s) for -: 'list' and 'int'
>>>

示例展示了正常的加减法调用，以及当参数类型不匹配时，服务端抛出的 TypeError 如何被正确传递并显示在客户端。

安全性分析与 JSON 替代方案

上述基于 pickle 的实现虽然简单，但存在严重的安全隐患。pickle 在反序列化时可能执行任意代码。如果攻击者能够构造特定的序列化消息发送给服务器，即可在服务端执行任意函数，导致服务器被完全控制。因此，绝对不要允许来自不信任或未认证客户端的 RPC 请求，更不允许从 Internet 直接暴露 RPC 服务。该方案仅适用于位于防火墙内部、受信任的内部网络环境。

使用 JSON 替代 pickle

为了消除 pickle 带来的安全风险，可以使用 JSON、XML 或其他安全的数据交换格式替代 pickle。以下是基于 JSON 的 RPC 实现示例：

# jsonrpcserver.py
import json

class RPCHandler:
  def __init__(self):
    self._functions = { }

  def register_function(self, func):
    self._functions[func.__name__] = func

  def handle_connection(self, connection):
    try:
      while True:
        # Receive a message
        func_name, args, kwargs = json.loads(connection.recv())
        # Run the RPC and send a response
        try:
          r = self._functions[func_name](*args,**kwargs)
          connection.send(json.dumps(r))
        except Exception as e:
          connection.send(json.dumps(str(e)))
    except EOFError:
       pass

# jsonrpcclient.py
import json

class RPCProxy:
  def __init__(self, connection):
    self._connection = connection
  def __getattr__(self, name):
    def do_rpc(*args, **kwargs):
      self._connection.send(json.dumps((name, args, kwargs)))
      result = json.loads(self._connection.recv())
      return result
    return do_rpc

在 JSON 方案中，pickle.loads() 被替换为 json.loads()，pickle.dumps() 被替换为 json.dumps()。需要注意的是，由于 JSON 无法直接序列化 Python 异常对象，服务端在捕获异常时，需将异常转换为字符串（str(e)）后再发送。客户端在接收响应时，若需处理异常，需自行解析该字符串并构造相应的异常对象抛出。

总结

通过 multiprocessing 库，我们可以快速构建一个功能完整的远程过程调用系统。虽然 pickle 方案实现简单，但出于安全考虑，建议在内部网络环境中使用，或采用 JSON 等安全的序列化格式。在实际部署中，务必结合身份验证（如 authkey）和防火墙策略，确保 RPC 服务不被未授权访问。

以上就是基于 Python multiprocessing 实现远程方法调用的详细内容，更多关于 Python 网络编程与 RPC 技术的资料请关注本站其它相关文章！