如何在Python独立进程间实现双向跨进程函数调用？

原代码报错原因 #

你用multiprocessing.BaseManager传递函数时，客户端拿到的func是远程对象的代理，并非真正可直接调用的函数。BaseManager的设计更偏向父子进程或由管理器启动的进程场景，独立进程之间共享函数代理会因为序列化上下文缺失，导致代理无法正确解析参数并触发远程调用。

原生Python实现方式 #

原生可以通过multiprocessing.connection结合pickle手动实现双向IPC，但需要自己处理：

• 消息的序列化与反序列化（注意pickle的安全风险，仅在可信环境使用）
• 回调函数的注册、注销逻辑
• 连接状态检测（通过心跳或异常捕获感知进程终止）
  这种方式需要编写大量底层代码，不够简洁，Pythonic程度低。

推荐的第三方包 #

RPyC #

这是最符合你需求的工具，支持双向RPC、动态注册回调、连接状态感知，完全Pythonic，无需复杂配置。

示例代码

服务器端（支持客户端注册回调，也可主动调用客户端函数）：

import rpyc
from rpyc.utils.server import ThreadedServer

class MyService(rpyc.Service):
    def on_connect(self, conn):
        # 客户端连接时记录连接对象，用于主动调用
        self.client_conn = conn
        print("客户端已连接")

    def on_disconnect(self, conn):
        print("客户端已断开")

    def exposed_server_func(self, msg):
        # 暴露给客户端调用的函数
        print(f"服务器收到消息：{msg}")
        return f"服务器响应：{msg}"

    def call_client_func(self):
        # 主动调用客户端注册的函数
        if hasattr(self.client_conn.root, "client_func"):
            result = self.client_conn.root.client_func("来自服务器的调用")
            print(f"客户端返回：{result}")

if __name__ == "__main__":
    server = ThreadedServer(MyService, port=50000)
    # 单独线程触发主动调用，简化为输入触发
    import threading
    def trigger_call():
        while True:
            input("按回车调用客户端函数：")
            server.service.call_client_func()
    threading.Thread(target=trigger_call, daemon=True).start()
    server.start()

客户端（注册回调，可调用服务器函数，感知连接断开）：

import rpyc

def client_func(msg):
    print(f"客户端收到服务器调用：{msg}")
    return f"客户端响应：{msg}"

if __name__ == "__main__":
    try:
        conn = rpyc.connect("localhost", 50000)
        # 向服务器注册客户端函数
        conn.root.client_func = client_func
        # 调用服务器函数
        response = conn.root.server_func("来自客户端的消息")
        print(response)
        # 保持连接，等待服务器主动调用
        while True:
            pass
    except ConnectionRefusedError:
        print("无法连接到服务器")
    except EOFError:
        print("服务器已终止连接")

特点 #

• 双向RPC：服务器和客户端可互相调用对方暴露/注册的函数
• 动态注册：无需预定义所有函数，运行时可注册、注销回调
• 连接感知：通过异常（如EOFError）捕获对方进程终止
• 简洁易用：API设计Pythonic，无需处理底层套接字细节

内容的提问来源于stack exchange，提问作者mo FEAR