Python使用rpyc库实现RPC远程调用示例

RPC核心原理与rpyc组件作用

远程过程调用（RPC）允许当前进程像调用本地函数一样触发其他进程的代码。实现该机制通常需要完成三件事：提取待调用的接口签名、将参数序列化并通过网络传输、对侧接收反序列化后执行并回传结果。rpyc 正是专注于简化这一流程的开源库，它内置了对象序列化、TCP Socket通信管理以及动态代理生成能力，开发者只需关注业务逻辑即可。

服务端构建与端口监听

第1步：定义并注册暴露的服务接口

目的：声明哪些方法允许被外部网络请求访问。
入口/动作：新建类继承自 rpyc.Service，并在目标方法名前添加 exposed_ 前缀。
重要参数：方法必须包含 self 作为首参，参数类型需保持可序列化（如字典、列表、基础类型）。
完成状态：服务端实例化后，仅带有 exposed_ 前缀的方法会通过反射机制映射为远程可用接口，未标记的方法默认隔离于内部私有域。

from rpyc import Service

class RemoteService(Service):
    # 客户端连接成功时的回调，可用于记录初始日志
    def on_connect(self, conn):
        print("New client connected")

    '''供客户端调用的方法前必须加 exposed_'''
    def exposed_search(self, data):
        # 模拟数据处理逻辑
        data['status'] = 'processed'
        data['result'] = 'ok'
        return data

第2步：启动多线程网络服务器

目的：绑定本地IP与特定端口，开启后台线程池持续监听 incoming 连接。
入口/动作：导入 rpyc.utils.server.ThreadedServer，传入自定义的 Service 类并调用 .start()。
重要参数：hostname='127.0.0.1' 限定本地回环；port 指定通信端口；auto_register=False 禁用Windows等系统下不必要的防火墙自动放行弹窗，提升安全性。
完成状态：终端打印 Server IP: 及端口号，进程进入阻塞态，随时接受客户端握手请求。

from rpyc.utils.server import ThreadedServer

def start_server(port):
    sr = ThreadedServer(
        service=RemoteService,
        hostname='127.0.0.1',
        port=port,
        auto_register=False
    )
    print(f'Server listening at port: {sr.port}')
    sr.start()

# 使用多进程池并发启动两个独立服务端口用于压测对比
if __name__ == '__main__':
    from multiprocessing import Pool
    pool = Pool(processes=2)
    pool.apply_async(start_server, args=(9998,))
    pool.apply_async(start_server, kwds={'port': 9999})
    pool.close()
    pool.join()

客户端连接与数据交互

第3步：发起客户端远程调用请求

目的：向已知的服务端地址发送结构化数据，并等待远端执行完毕后接收返回值。
入口/动作：通过 rpyc.connect(host, port) 建立长连接通道；通过 conn.root. 对象调用目标方法。
重要参数：连接时会自动附加 threaded_config 确保并发安全；调用方法时无需书写 exposed_ 前缀，rpyc 会在底层代理层自动完成名称映射。
完成状态：控制台输出服务端回传的完整字典结构，手动执行 conn.close() 切断网络引用以释放连接池。

import rpyc
import time

def query_remote_data(query, port):
    # 建立与服务端的连接通道
    conn = rpyc.connect('localhost', port)
    try:
        # 调用时省略 exposed_ 前缀，rpyc 会自动路由
        result = conn.root.search(query)
        time.sleep(1)  # 模拟业务耗时
        print(f'Response from port {port}: {result}')
    finally:
        # 务必关闭连接，防止文件句柄耗尽
        conn.close()

if __name__ == '__main__':
    from multiprocessing import Pool
    payload = {"match": {"name": "user"}}
    pool = Pool(processes=2)
    pool.apply_async(query_remote_data, args=(payload, 9998))
    pool.apply_async(query_remote_data, kwds={'query': payload, 'port': 9999})
    pool.close()
    pool.join()

运行验证与注意事项

启动服务端脚本后，客户端将在控制台中按顺序打印出端口 9998 和 9999 的处理结果。两者返回的字典结构完全一致，证明 rpyc 能够正确处理复杂对象的网络透传。在实际项目中若遇到连接超时（Connection Refused），请检查目标端口是否被防火墙拦截或服务端是否已完成初始化。涉及修改生产环境暴露方法时，建议先评估 on_disconnect 清理逻辑，防止内存泄漏。

以上就是使用rpyc在Python中实现RPC远程过程调用的详细内容，更多关于Python网络编程与分布式接口开发的资料请关注本站其它相关文章！