简介面向 Unity3D 与 Python 开发者的跨语言进程间通信示例包基于 ZeroMQ 网络库实现 C# 与 Python 之间的高速、轻量、通用数据交换既适合需要实时传递图像、视频、文本、JSON 或文件数据的 Unity 项目也可作为入门跨语言通信的参考范本。压缩包共 51 个文件大小约 827KB包含 Unity 工程资源asset/meta/unity、C# 脚本、Python 服务端脚本、netmq 与 pyzmq 依赖的 DLL以及 json/xml 配置、README 说明等目录划分清楚便于按模块查看。示例提供可直接对照的 server.py 和 Unity C# 端工程内置零配置启动逻辑使用者不需要显式绑定套接字或了解底层网络细节即可快速搭建通信原型同时附带 troubleshooting 排错截图与命令行运行演示 GIF对理解 PyZMQ 和 NetMQ 的对接流程、定位配置问题很有帮助目录组织方式也便于迁移到自己的项目结构。目前已有 1717 人学习下载适合希望在 Unity 项目中低成本接入 Python 服务的中高级开发者快速上手。1. Unity3D 与 Python 的进程间通信为什么 ZeroMQ 是那个最省心的答案做 Unity3D 与 Python 联调的时候最烦的不是写业务逻辑而是两个进程怎么把数据递过去。Python 那边跑深度学习推理、仿真计算Unity 这边出画面、处理交互两边用 C# 和 Python 各有各的生态硬凑在一起写文件轮询或原始 Socket 都能用但动不动就卡、丢、崩。我拆过的这个 Unity3D-Python-Communication 示例核心是用 ZeroMQ 在 C# 和 Python 之间搭一条快速、简单、通用的进程间通信链路Python 端用 pyzmqUnity 端用 NetMQ两边代码都很短却能撑住高频帧数据的推拉。适合要做仿真数据可视化、AI 推理结果回传、多进程联调的开发者也适合刚接触进程间通信的人拿它当第一份可跑的范例。2. 通信方案选型逻辑raw socket、HTTP 与 ZeroMQ 的取舍2.1 为什么不是原始 TCP两条血泪场景很多人第一反应是直接用 C# 的TcpClient和 Python 的socket写。这种方式在双方都在本机、一次只传一条消息时没问题但两个场景很容易翻车。第一个是消息边界TCP 是流协议发端Send了三次收端可能一次Receive就把三条都读出来也可能一条分成两半。你不得不在业务里自己搞长度前缀或者分隔符这个看起来很简单的“加个包头”工作在 Unity 的帧循环里接收时就容易出玄学 bug。第二个场景更痛断线重连。Unity 的编辑器在播放和停止之间反复切换Python 脚本也经常 CtrlC 重启。raw socket 的客户端断线后要自己处理重连、半包、粘包、对端关闭握手稍有一个时序没兜住进程就直接崩。这些代码写起来不短调试起来还难复现。所以这个示例没有走 raw socket而是直接用 ZeroMQ让库去处理连接建立、断线重连和消息分帧。2.2 ZeroMQ 的抽象层次socket 类型而不是连接ZeroMQ 和普通 socket 最大的不同是它不面向“连接”而面向“消息队列”。你不需要在代码里维护一条连接的打开和关闭只需要创建 socket、调用 bind 或 connect、发送或接收消息。连接的重建、排队、丢包策略都由库里层处理。常见的 socket 类型里这个通信链条上最核心的是下面这张表模式方向典型场景PUB/SUB一对多广播Python 推仿真数据Unity 订阅显示REQ/REP一问一答Unity 请求推理结果Python 返回PUSH/PULL分发与汇集多个 Python 进程计算结果汇总PAIR一对一双向两个进程做对称交互在 Unity 和 Python 之间PUB/SUB 是用得最顺手的。Python 端作为 Publisher 推送带 topic 的消息Unity 端作为 Subscriber 订阅感兴趣的 topic双向解耦不需要知道对方在不在线。2.3 pyzmq 与 NetMQ 的绑定差异两边既然语言不同绑定的库也不一样。Python 端用 pyzmq这是对零库的 Python 绑定API 稳定pip install pyzmq就能用。Unity C# 端用 NetMQ它是一套纯 C# 的 ZeroMQ 移植实现不依赖 C 原生库装进 Unity 工程里不需要额外处理 DLL 的原生依赖。这里有个容易忽略的差异pyzmq 是原生库绑定性能和重连策略完全按标准实现走NetMQ 是纯托管实现功能覆盖了大部分常用 socket 类型但某些高级 IO 细节和 pyzmq 不是完全一致。实际使用中不要把两边的高级参数当成同一套东西来配最省事的做法是把通信内容限制在“多帧消息加字符串”这个层面能避开绝大多数差异。3. Python 端先跑起来pyzmq 的 PUB/SUB 最小实现3.1 环境准备先装 pyzmq 并验证绑定Python 端只需要一个依赖。创建虚拟环境后执行pip install pyzmq python -c import zmq; print(zmq.__version__)能打印出版本号就说明绑定没问题。这一步卡住的常见情况是虚拟环境没激活装到了全局环境里如果是在公司内网机器上还需要确认 pip 的镜像源可访问。注意 pyzmq 对 Python 版本要求比较宽3.8 到 3.12 都能正常安装不用刻意追新版本。3.2 写一个 Publisher发送带有帧号的数据下面这段代码是数据发送的最小骨架。这里我用它模拟推流一帧骨骼数据或传感器数据import zmq import time import json context zmq.Context() socket context.socket(zmq.PUB) # 关键参数发送端高水位线默认 1000这里提到 10000 socket.setsockopt(zmq.SNDHWM, 10000) # linger 设置为 0关闭时立即丢弃未发送数据避免进程退出卡住 socket.setsockopt(zmq.LINGER, 0) socket.bind(tcp://*:5555) print(Publisher 启动监听 tcp://*:5555) frame_id 0 while True: payload { frame_id: frame_id, timestamp: time.time(), data: [0.1 * frame_id, 0.2 * frame_id, 0.3 * frame_id], } # topic 用 skeleton 前缀订阅端可以按前缀过滤 socket.send_string(skeleton json.dumps(payload)) frame_id 1 time.sleep(0.01)这里send_string发送的是主题加正文合并的一条消息主题和内容之间用空格分隔是我自己定的协议订阅端按空格拆分即可。SNDHWM表示发送队列里最多攒多少条未发送消息超过后库会按丢弃策略处理LINGER是最容易踩的坑默认 -1 代表关闭 socket 时无限等待把队列发完这在循环里按 CtrlC 时会让进程干等很久设为 0 就能立即退出代价是可能丢弃尾部几条数据。3.3 写一个 Subscriber订阅前缀并做校验对应地接收端代码长这样import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.setsockopt(zmq.RCVHWM, 10000) socket.connect(tcp://127.0.0.1:5555) # 订阅空字符串代表接收所有 topic socket.setsockopt_string(zmq.SUBSCRIBE, skeleton) while True: try: message socket.recv_string(flagszmq.NOBLOCK) topic, payload message.split( , 1) print(f收到 topic[{topic}] 数据{payload[:80]}) except zmq.Again: # 没有新消息时跳过继续循环 pass缺失订阅过滤导致收不到数据是最常见的新手翻车点ZMQ 的 SUB 端默认不接收任何消息必须通过SUBSCRIBE设置订阅前缀。这里订阅skeleton就只收这个主题订阅空字符串才收全部。NOBLOCK标志让接收变为非阻塞配合 try/except 做轮询。这样做的好处是接收循环不会卡在recv_string()上方便后续在这个循环里叠加心跳超时判断。3.4 bind 与 connect顺序错了反而丢消息bind 和 connect 的规则和普通 socket 不同。ZMQ 里bind 方不一定要先启动connect 方即使连的是还不存在的端口也“成功”连接会在对端 bind 之后自动建立。做不到的是数据保证PUB 在 SUB 完成连接之前发出的消息是会被丢弃的。所以在这个链路里我的习惯是让 Unity 的 Subscriber 先 bind 到一个端口Python 的 Publisher 后启动并 connect 过去。这样 Python 一启动数据就源源不断推到 Unity不存在“Unity 还没准备好就开推”的窗口期。反过来如果 Python 先 bind、Unity 后 connectPython 这边一启动就狂发消息而这些消息都会丢——因为订阅端还没建立连接。4. Unity C# 端接入NetMQ 的异步接收与生命周期管理4.1 Unity 的帧循环和 NetMQ 的线程模型Unity 的 MonoBehaviour 逻辑跑在主线程而 NetMQ 的接收是走自己的线程的。不能直接在Update()里调用阻塞 Receive否则一帧卡住几毫秒就丢了交互帧率。也不要把 NetMQ 的事件线程直接引到 Unity 的 UI 线程上访问 GameObjectUnity 的 API 只能在主线程操作。正确的做法是让 NetMQ 在自己的线程收数据收完丢进一个线程安全的队列然后在Update()里取出队列内容做真正的游戏逻辑处理。4.2 接收器实现Poller 加队列下面这段源码是这个示例里最值得直接抄走的部分using NetMQ; using NetMQ.Sockets; using System.Collections.Concurrent; using UnityEngine; public class ZMQReceiver : MonoBehaviour { private SubscriberSocket subscriber; private NetMQPoller poller; private readonly ConcurrentQueuestring messageQueue new ConcurrentQueuestring(); void Start() { // 在子线程跑 NetMQ 的 Poller不占用主线程 subscriber new SubscriberSocket(); subscriber.Options.ReceiveHighWaterMark 10000; subscriber.Bind(tcp://*:5556); subscriber.Subscribe(skeleton); poller new NetMQPoller { subscriber }; subscriber.ReceiveReady (s, e) { string message e.Socket.ReceiveFrameString(); messageQueue.Enqueue(message); }; poller.RunAsync(); } void Update() { // 主线程只做出队和业务处理 while (messageQueue.TryDequeue(out string message)) { Debug.Log(收到数据: message.Substring(0, Mathf.Min(message.Length, 80))); } } void OnDestroy() { poller?.Stop(); poller?.Dispose(); subscriber?.Dispose(); } }NetMQPoller是这个链路里的关键角色。它管理 socket 的事件分发RunAsync()会让 poller 在后台线程跑socket 收到数据后触发ReceiveReady回调。回调里ReceiveFrameString()取出一条消息ReceiveFrameString对应 Python 端send_string发出来的一帧。消息放入ConcurrentQueue后立即返回不做解析把耗时操作留到主线程处理。这里还用ReceiveHighWaterMark设置了接收队列长度防止 Python 推太快时 Unity 这边内存无上限增长。注意OnDestroy里必须做清理。Unity 编辑器在停止播放时会销毁这个组件如果 poller 没有停掉它会一直占着端口和线程下次再点 Play 就会出现“端口被占用”或者进程残留。4.3 发送数据从 Unity 反向推给 Python双向通信也很自然。Unity 端创建一个 PublisherSocketPython 端开一个 Subscriber 连过来即可using NetMQ; using NetMQ.Sockets; using UnityEngine; public class ZMQPublisher : MonoBehaviour { private PublisherSocket publisher; void Start() { publisher new PublisherSocket(); publisher.Options.SendHighWaterMark 5000; publisher.Options.Linger System.TimeSpan.Zero; publisher.Connect(tcp://127.0.0.1:5557); } public void SendFrame(int frameId, Vector3 position) { string payload ${frameId},{position.x:F3},{position.y:F3},{position.z:F3}; publisher.SendFrame(pose payload); } void OnDestroy() { publisher.Dispose(); } }这里把Linger设为TimeSpan.Zero原因是 Unity 编辑器频繁切换播放模式不设置的话 Dispose 时可能等待发送队列清空而卡顿。SendFrame组装字符串时用简单逗号分隔配合 Python 端的 split 解析速度快且不会因为 JSON 序列化引入额外开销。4.4 请求应答模式Unity 需要结果时用 REQ/REP如果 Unity 端的动作需要等 Python 返回一个结果比如“这一帧的识别结果是什么”PUB/SUB 就不合适了。这时用 REQ/REP 模式using NetMQ; using NetMQ.Sockets; public static class ZMQRequest { public static string Request(string address, string question) { using (var req new RequestSocket()) { req.Options.Linger System.TimeSpan.FromSeconds(2); req.Connect(address); req.SendFrame(question); if (req.TryReceiveFrameString(System.TimeSpan.FromSeconds(5), out string answer)) { return answer; } return timeout; } } }这个静态方法每次请求都现建 socket用完即关。它适合低频调用比如用户点击一次 UI。TryReceiveFrameString的第二个参数是超时时间防止 Python 进程挂死导致 Unity 端一直等下去。这个模式下的 Python 端就是标准的 REP 循环收到什么回什么。5. 避坑记录从连不上到崩溃的五个踩坑现场5.1 现象Unity 停止播放后Python 端进程还挂着第一次跑通 pub/sub 开始联调了结果我点 Unity 的停止按钮Python 端那边消息照常打印完全没感知到对端已经没了。原因ZMQ 的 PUB 端根本不管 SUB 端是否在线照常往队列里发。解决我在 Python 的发送循环里加了订阅数量检测或者更简单——直接发心跳如果连续五秒没有收到任何 ACK用 PAIR 或额外的 REP 通道就主动断开退出。5.2 现象消息丢失看起来像网络问题订阅端明明连上了数据就是少。排查了半天抓包也没找到 TCP 重传。原因是 PUB/SUB 模式下如果 SUB 端的处理速度跟不上 PUB 的发送速度ZMQ 会按高水位线丢消息而且不会通知应用层。解决调大两端 HWM同时让 Unity 端接收线程只做入队别在回调里做任何耗时操作如果业务上就是处理不过来就在 Publisher 端稍微 sleep 一下控制发送频率。5.3 现象Connect 返回成功却收不到数据Unity 端Connect(tcp://localhost:5555)没有报错但消息就是不来。这种玄学大多出在两个点上。第一个是 Python 端 bind 的是tcp://*:5555Unity 端却写成了5556端口对不上。第二个是 SUB 端没有调Subscribe前面说过不订阅就收不到任何消息。先把这两个点排查掉再谈网络问题。5.4 现象Unity 编辑器卡死Debug.Log 刷得飞起把 NetMQ 的 Receive 直接写在Update()里跑了结果一启动画面就冻住。原因ReceiveFrameString()是阻塞方法没有消息时它会卡住主线程Unity 整个 UI 就假死。解决一定用 Poller 配合后台线程或者用TryReceiveFrameString(TimeSpan.Zero)非阻塞轮询。从那以后我定了条规矩任何网络代码都不允许直接出现在 Update、FixedUpdate 这些帧回调里。5.5 现象跑了几分钟后内存占用匀速上涨PUB/SUB 正常跑了一会儿任务管理器里 Python 的内存一条直线往上走。原因订阅端处理不过来消息全积压在 RCVHWM 队列里而我把 RCVHWM 调到了十万积压几万条 JSON 字符串每个都是几十 KB内存很快就吃满了。解决把 HWM 调回一个合理值比如 1000 到 5000同时监控积压情况。积压了就让 Unity 端降级处理比如合并最近几帧的数据而不是一条一条全处理。6. 进阶验证压测脚本、HWM 参数与消息压缩的取舍6.1 一个可重复的压测循环要把通信链路调稳不能靠肉眼得写个脚本记录“发送了多少条、收到多少条、延迟多少”。我这里用一个 Python 脚本做压测接收端import zmq import time ctx zmq.Context() sock ctx.socket(zmq.SUB) sock.setsockopt(zmq.RCVHWM, 50000) sock.connect(tcp://127.0.0.1:5558) sock.setsockopt_string(zmq.SUBSCRIBE, test) expected 0 received 0 lost_count 0 start time.time() duration 5.0 while time.time() - start duration: try: frame sock.recv_string(flagszmq.NOBLOCK) seq int(frame.split( , 1)[1]) if seq ! expected: lost_count seq - expected expected seq 1 received 1 except zmq.Again: pass elapsed time.time() - start print(f实际接收 {received} 条理论应到 {expected} 条丢包 {lost_count} 条) print(f吞吐率 {received / elapsed:.0f} 条/秒)这个脚本通过帧序号检测丢包。配合 Unity 端用 Publisher 发递增序号能很快评估这条链路的吞吐边界也能量化 HWM 参数调整的前后效果。6.2 HWM 参数的配合调节HWM 设置不是越大越好。PUB 端 SNDHWM 和 SUB 端 RCVHWM 是独立生效的任何一端达到上限就开始丢。我的经验是把两端设成一致然后从 1000 往上试探。先压测拿到当前场景的每秒消息数再倒推 HWM 设多少够用。比如一秒钟推 100 条、Unity 端每帧处理一条那么 HWM 设 500 到 1000 就足够应对几帧的瞬时积压设五万反而掩盖了业务处理不过来的问题。6.3 消息压缩什么时候值得做消息体比较大的时候比如一帧传完整点云或深度图字符串直传会把网卡和内存都打满。这种压缩就值得做发送端先 gzip接收端解压。但压缩是用 CPU 换带宽数据小于 1KB 时压缩收益很低反而增加延迟。我一般会把阈值定在 4KB 以上才考虑压缩并且在代码里做一个开关方便在压测时对比。从那以后我每次搭 Unity 和 Python 的新链路都强制走一遍这个流程先跑通最小 pub/sub然后写一个带帧序号的压测脚本最后再进业务逻辑。这套顺序帮我省掉了太多排查时间希望也能帮到你。本文还有配套的精品资源点击获取
