VeighNa RpcService 模块深度指南基于 ZeroMQ 的多进程分布式交易路由【免费下载链接】vnpy基于Python的开源量化交易平台开发框架项目地址: https://gitcode.com/gh_mirrors/vn/vnpyRpcService 是 VeighNa Trader 中用于将单个交易进程转化为 RPC 服务器的功能模块对外提供交易路由、行情数据推送、持仓资金查询等服务。本文以 rpc_service.md 为骨架结合仓库中 vnpy/rpc 的底层实现与 examples/client_server 的完整示例讲解从服务端配置、客户端接入到源码级通讯机制的全过程读完即可搭建一套一条通道、多客户端并行交易的分布式交易架构。功能简介RpcService 的核心作用是把一个 VeighNa Trader 进程升级为RPC 服务器服务端对外提供两类能力交易路由客户端进程通过 RPC 远程调用服务端的下单、撤单等交易接口数据分发服务端将收到的行情、委托、成交等事件数据主动推送给所有已连接的客户端。服务端集中连接真实的交易接口如 CTP客户端无需再次配置账户密码只需与服务端通讯即可获得与本地直连几乎一致的使用体验。关于 RPC 更宏观的应用场景见本文末尾【RPC 的应用场景】一节。加载启动通过 VeighNa Station 加载启动并登录 VeighNa Station 后点击【交易】按钮进入 VeighNa Trader在配置对话框的【应用模块】栏勾选【RpcService】即可。通过脚本加载在启动脚本中按以下方式添加应用模块# 写在顶部 from vnpy_rpcservice import RpcServiceApp # 写在创建main_engine对象后 main_engine.add_app(RpcServiceApp)仓库自带的 examples/client_server/run_server.py 提供了完整的服务端启动脚本除加载RpcServiceApp外还演示了如何在无 GUI 的命令行模式下启动 RPC 引擎from vnpy_rpcservice import RpcServiceApp from vnpy_rpcservice.rpc_service.engine import RpcEngine, EVENT_RPC_LOG rpc_engine: RpcEngine main_engine.add_app(RpcServiceApp) # 连接交易接口以 CTP 为例 main_engine.connect(setting, CTP) sleep(10) rep_address: str tcp://127.0.0.1:2014 pub_address: str tcp://127.0.0.1:4102 rpc_engine.start(rep_address, pub_address)从 examples/veighna_trader/run.py 可以看到RpcServiceApp 也以注释形式作为可选应用模块预置在标准启动脚本中取消注释即可启用。启动模块启动 RPC 服务模块之前请先连接登录交易接口连接方法见基本使用篇的连接接口部分。正确连接后VeighNa Trader 主界面【日志】栏会输出合约信息查询成功此时再启动 RPC 模块可确保客户端接入后能立即查询到合约、持仓、资金等初始化信息。连接交易接口成功后通过菜单栏【功能】-【RPC 服务】或点击左侧按钮栏的图标即可进入 RPC 服务模块的 UI 界面。配置与使用配置 RPC 服务RPC 服务基于ZeroMQ开发对外提供两个通讯地址职责各不相同请求响应地址REPRequest-Reply 模式用于被动接收客户端发送过来的请求执行对应任务后返回结果典型功能举例行情订阅委托下单委托撤单初始化信息查询合约、持仓、资金等。事件广播地址PUBPublish-Subscribe 模式用于主动推送服务端收到的事件数据到所有已连接的客户端典型功能举例行情推送委托推送成交推送。两个地址均采用 ZeroMQ 的地址格式由通讯协议如tcp://和通讯地址如127.0.0.1:2014两部分组成。RPC 服务支持的通讯协议如下协议前缀适用系统通讯范围说明TCP 协议tcp://Windows 和 Linux 均可本机127.0.0.1或网络局域网 IP通用推荐默认使用IPC 协议ipc://仅 LinuxPOSIX 本地端口通讯仅限本机后缀为任意字符串内容低延时场景使用一般推荐直接使用 TCP 协议及默认地址对于使用 Ubuntu 系统、追求更低通讯延时的用户可以改用 IPC 协议。运行 RPC 服务完成通讯地址配置后点击【启动】按钮即可启动 RPC 服务日志区域会输出RPC 服务启动成功。启动成功后即可在另一个 VeighNa Trader 进程客户端中使用 RpcGateway 连接。如需停止服务点击【停止】按钮此时日志输出RPC 服务已停止。连接客户端VeighNa 提供了与 RpcService 配套使用的RpcGateway作为客户端的标准接口来连接服务端并进行交易对上层应用透明。从客户端的视角看RpcGateway 是一个类似 CTP 的接口——因为服务端已经统一完成了外部交易账户的配置与连接客户端只需与服务器端通讯无需再次输入账户密码等信息。在客户端加载 RpcGateway 接口后进入 VeighNa Trader 主界面点击菜单栏【系统】-【连接 RPC】在弹出的窗口中点击【连接】即可使用。窗口中【主动请求地址】和【推送订阅地址】分别对应服务端配置的【请求响应地址】和【事件广播地址】注意不要写反。仓库的 examples/client_server/run_client.py 展示了客户端脚本的标准写法加载RpcGateway作为交易接口再叠加策略应用如 CtaStrategyAppfrom vnpy_rpcservice import RpcGateway from vnpy_ctastrategy import CtaStrategyApp main_engine.add_gateway(RpcGateway) main_engine.add_app(CtaStrategyApp)RPC 简介由于全局解释器锁 GIL 的存在单一 Python 进程只能利用 CPU 单核的算力。远程过程调用Remote Procedure Call Protocol, RPC服务可以用于跨进程或者跨网络的服务功能调用有效解决了上述问题由一个特定进程连接交易接口充当服务器角色在本地物理机或局域网内主动向其他独立的客户端进程推送事件并处理客户端发来的相关请求。源码剖析RPC 底层通讯机制理解了界面操作后再深入仓库源码可以看到 RPC 服务的完整实现位于 vnpy/rpc 目录共三个文件common.py、server.py、client.py。通用配置心跳参数vnpy/rpc/common.py 定义了服务端与客户端共用的心跳常量HEARTBEAT_TOPIC heartbeat # 心跳消息主题 HEARTBEAT_INTERVAL 10 # 服务端心跳推送间隔秒 HEARTBEAT_TOLERANCE 30 # 客户端心跳容忍超时秒该文件还通过signal.signal(signal.SIGINT, signal.SIG_DFL)恢复 Ctrl-C 中断的默认行为保证 RPC 接收线程可被正常打断退出。服务端RpcServervnpy/rpc/server.py 中的RpcServer是服务端核心类其构造函数创建了两种 ZeroMQ Socket_socket_rep类型zmq.REP请求-应答模式服务端 Socket对应【请求响应地址】_socket_pub类型zmq.PUB发布-订阅模式服务端 Socket对应【事件广播地址】。核心方法包括start(rep_address, pub_address)对两个地址执行bind绑定随后启动独立的工作线程threading.Thread(targetself.run)并初始化下一次心跳推送时间。register(func)以函数名func.__name__为键将可调用对象注册进_functions字典。RpcService 应用正是通过这一机制把行情查询、下单、撤单等引擎方法批量注册为可被客户端远程调用的函数。run()工作线程主循环。先对 REP Socket 轮询 1 秒期间调用check_heartbeat()收到请求后通过recv_pyobj()反序列化出(函数名, 位置参数, 关键字参数)从_functions中取出对应函数执行并将结果打包为[True, 返回值]或异常信息[False, traceback]后send_pyobj()回传。publish(topic, data)加锁后向 PUB Socket 推送[topic, data]格式的数据帧。check_heartbeat()每HEARTBEAT_INTERVAL10 秒向heartbeat主题推送一次当前时间戳供客户端判断连接存活状态。客户端RpcClientvnpy/rpc/client.py 中的RpcClient是客户端核心类构造时创建_socket_reqzmq.REQ请求-应答模式与_socket_subzmq.SUB发布-订阅模式并设置TCP Keepalive参数zmq.TCP_KEEPALIVE1、空闲 60 秒用于检测异常断连。其最精妙的设计是__getattr__动态代理配合lru_cache(100)缓存def __getattr__(self, name: str) - Any: def dorpc(*args: Any, **kwargs: Any) - Any: timeout: int kwargs.pop(timeout, 30000) # 默认超时 30 秒 req: list [name, args, kwargs] with self._lock: self._socket_req.send_pyobj(req) n: int self._socket_req.poll(timeout) if not n: raise RemoteException(fTimeout of {timeout}ms reached for {req}) rep self._socket_req.recv_pyobj() if rep[0]: return rep[1] else: raise RemoteException(rep[1]) return dorpc也就是说只要服务端注册了某函数客户端就能以同名属性直接调用形如client.add(1, 3)即可触发一次远程调用默认请求超时为 30000 毫秒可通过关键字参数timeout覆盖。调用失败或超时时抛出RemoteException其异常信息包含服务端traceback.format_exc()捕获的完整堆栈便于排查服务端执行错误。RpcClient的运行线程run()持续轮询 SUB Socket一旦超过HEARTBEAT_TOLERANCE30 秒未收到服务端心跳便调用on_disconnected()打印RpcServer has no response...告警收到heartbeat主题时刷新_last_received_ping收到业务主题时交给用户覆写的callback(topic, data)处理。订阅主题通过subscribe_topic(topic)完成传入空字符串表示订阅全部主题。最小可运行示例simple_rpc仓库在 examples/simple_rpc 中提供了不依赖 GUI 的最小演示可用于快速理解 RPC 的本质test_server.py继承RpcServer在构造函数中register(self.add)注册远程函数绑定tcp://*:2014请求响应与tcp://*:4102事件广播启动后每 2 秒向test主题发布一次服务器时间test_client.py继承RpcClient覆写callback()打印收到的主题与数据连接tcp://localhost:2014与tcp://localhost:4102并循环执行tc.add(1, 3)验证远程调用。注意这里的端口号2014/4102与 examples/client_server/run_server.py 中使用的默认地址tcp://127.0.0.1:2014/tcp://127.0.0.1:4102保持一致是 VeighNa RPC 生态约定俗成的默认端口日常使用可以沿用。RPC 服务RpcService的应用场景策略数量较多的个人用户只需本地一条行情和交易通道即可支持多个客户端进程同时交易且每个客户端中的交易策略独立运行、互不影响中小型投资机构用户可以在服务端加载各种交易接口以及 RiskManagerApp实现一个轻量级的资管交易系统多个交易员共享统一的交易通道并实现基金产品级别的风险管理。从 docs/community/info/introduction.md 对项目架构的描述看RpcService 被定位为允许将某一 VeighNa Trader 进程启动为服务端作为统一的行情和交易路由通道允许多客户端同时连接实现多进程分布式系统是 VeighNa 构建分布式量化交易架构的关键组件配合 vnpy_rpcservice 中的 RpcGateway客户端侧甚至可以实现无本地账户、纯远程路由的轻量接入模式。小结本文从功能定位、加载启动、地址配置、客户端接入四个层面完整介绍了 RpcService 的使用方法并通过 vnpy/rpc 源码剖析了其基于 ZeroMQ REP/REQ 与 PUB/SUB 双通道的通讯模型、动态函数注册与远程代理、10 秒心跳 30 秒容忍的超时机制等实现细节。结合 examples/client_server 与 examples/simple_rpc 两个示例读者可以快速在自己的环境里搭建起一条通道、多客户端并行交易的分布式交易架构。【免费下载链接】vnpy基于Python的开源量化交易平台开发框架项目地址: https://gitcode.com/gh_mirrors/vn/vnpy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考