使用 NiceGUI 与 ZeroMQ PUSH/PULL 构建实时数据流可视化看板

使用 NiceGUI 与 ZeroMQ PUSH/PULL 构建实时数据流可视化看板

【免费下载链接】niceguiCreate web-based user interfaces with Python. The nice way.项目地址: https://gitcode.com/GitHub_Trending/ni/nicegui

本指南基于 examples/zeromq 示例，讲解如何把 ZeroMQ 消息队列与 NiceGUI 异步 Web 框架结合，实现"生产者推送数据 → 浏览器实时刷新曲线"的完整链路。读完本文，你将掌握zmq.asyncio与 NiceGUI 事件循环的集成方式、PUSH/PULL 套接字的配对使用、ui.line_plot的实时 push 更新，以及用Event对象在长生命周期任务与短生命周期 UI 之间安全传递数据的方法，可直接照搬搭建自己的实时监控看板。

一、示例架构：PUSH/PULL 与 asyncio 的邂逅

本示例构建了一套极简的实时数据流水线，包含两个独立进程：

• 发布者（publisher）：zmq-server.py 是一个独立的 Python 脚本，通过zmq.PUSH套接字持续向tcp://localhost:5555发送随机整数；
• 订阅端 / GUI 服务器：main.py 是 NiceGUI 应用，通过zmq.PULL套接字连接到同一地址，接收数据后驱动前端折线图实时刷新。

选择 PUSH/PULL 模式（即"管道模式"）的原因在于：PUSH 端将消息公平地分发给已连接的 PULL 端，天然适合"单生产者、多消费者"的流式场景。示例只在单机回环地址localhost:5555上通信，若需跨机器部署，把两端连接地址替换为实际 IP 或域名即可，套接字类型与收发逻辑无需改动。

该示例最值得学习的一点，正如 README 强调的：核心在于使用zmq.asyncio创建可在 asyncio 事件循环中运行的订阅者。NiceGUI 服务器本身构建在 asyncio 之上，因此所有阻塞式 I/O（包括 ZeroMQ 收消息）都必须以异步方式接入，否则会卡死整个 Web 服务。

二、环境准备

除 NiceGUI 常规依赖外，还需安装 ZeroMQ 的 Python 绑定。示例的 requirements.txt 明确给出了版本下限：

nicegui>=3.0 pyzmq >= 26.0

pyzmq即 Python 版 ZeroMQ 绑定，其中zmq.asyncio子模块正是从pyzmq提供。安装命令：

pip install -r examples/zeromq/requirements.txt

注意：本文所有运行命令均默认在仓库根目录下执行。

三、发布者（Publisher）实现解析

zmq-server.py 全文仅 20 余行，却完整演示了zmq.asyncio的异步发送模型：

#!/usr/bin/env python3 import asyncio import random import zmq import zmq.asyncio context = zmq.asyncio.Context() socket = context.socket(zmq.PUSH) socket.bind('tcp://localhost:5555') async def send_loop(): while True: number = random.randint(0, 100) print(f'Sending number {number}') await socket.send(str(number).encode('ascii')) await asyncio.sleep(0.1) asyncio.run(send_loop())

关键细节说明：

• zmq.asyncio.Context()：异步上下文，其产生的套接字都支持await socket.send(...)，与asyncio事件循环无缝配合；
• socket.bind(...)：发布者作为服务端绑定在tcp://localhost:5555，等待订阅端连接；
• 发送频率：asyncio.sleep(0.1)将数据速率控制在每秒约 10 条，保证前端曲线刷新节奏可视且不压垮浏览器；
• 消息格式：str(number).encode('ascii')把整数编码为 ASCII 字节流，接收端用float(data)再还原为数值——这也是跨语言、跨进程通信的通用做法（如需传输复杂结构可改用 JSON）。

由于发送循环是死循环，推荐以python zmq-server.py &的方式放入后台运行（详见"运行步骤"一节）。

四、NiceGUI 服务器端：异步接收与实时绘图

GUI 端 main.py 是本示例的重头戏，完整代码如下：

#!/usr/bin/env python3 from datetime import datetime import zmq import zmq.asyncio from nicegui import Event, app, ui number_received = Event() context = zmq.asyncio.Context() socket = context.socket(zmq.PULL) socket.connect('tcp://localhost:5555') poller = zmq.asyncio.Poller() poller.register(socket, zmq.POLLIN) @ui.page('/') def page(): line_plot = ui.line_plot(n=1, limit=100, figsize=(10, 4)) number_received.subscribe(lambda number: line_plot.push([datetime.now()], [[number]])) @app.on_startup async def read_loop() -> None: while not app.is_stopped: events = await poller.poll() if socket in dict(events): data = await socket.recv() number = float(data) print(f'Received number {number}') number_received.emit(number) ui.run()

整个程序可以拆成四个层次理解。

4.1 建立异步 PULL 套接字与轮询器

context = zmq.asyncio.Context() socket = context.socket(zmq.PULL) socket.connect('tcp://localhost:5555') poller = zmq.asyncio.Poller() poller.register(socket, zmq.POLLIN)

• 与发布者的bind对应，这里使用connect主动连接tcp://localhost:5555；
• zmq.asyncio.Poller提供事件驱动的多路复用：await poller.poll()会挂起当前协程，直到有注册的套接字可读，从而避免忙轮询浪费 CPU；
• zmq.POLLIN表示"关注可读事件"，这是示例演示的注册方式。

4.2 在启动钩子中运行接收循环

@app.on_startup async def read_loop() -> None: while not app.is_stopped: events = await poller.poll() if socket in dict(events): data = await socket.recv() number = float(data) print(f'Received number {number}') number_received.emit(number)

• @app.on_startup是 NiceGUI 提供的生命周期装饰器（见 nicegui/app/app.py 中on_startup的定义），注册的处理器可以是同步或异步函数，在 NiceGUI 启动/重启时执行；
• while not app.is_stopped借助app.is_stopped属性（同样是 app.py 暴露的运行状态）作为循环退出条件，保证应用关闭时协程能干净退出；
• await poller.poll()与await socket.recv()全程非阻塞，与 NiceGUI 的 asyncio 事件循环共生共存，这正是 README 强调的"使用zmq.asyncio库创建可运行于 asyncio 循环中的订阅者"的具体落地。

4.3 用 Event 解耦"数据到达"与"UI 更新"

number_received = Event()

Event是 NiceGUI 3.0 起提供的通用事件分发原语（实现在 nicegui/event.py），其官方设计意图正是"在代码的不同部分之间分发信息，尤其是从像数据模型这样的长生命周期对象到短生命周期的 UI"。它提供subscribe（订阅回调）、emit（触发事件、不等待回调完成）、call（触发并等待全部回调完成）、emitted（等待事件发生）等接口。

本示例的使用模式非常典型：

• 接收循环（长生命周期后台任务）拿到数据后调用number_received.emit(number)；
• UI 页面在构建时通过number_received.subscribe(...)注册回调，把新数值推进折线图。

从源码看（nicegui/event.py），subscribe还做了内存安全兜底：当在 UI 上下文中订阅时，默认会在对应客户端被删除时自动取消订阅（unsubscribe_on_delete），避免回调长期持有已销毁的 UI 引用造成泄漏。emit内部对每个回调采用"触发即忘"（fire-and-forget）策略，异步回调会被放入后台任务执行，并通过app.handle_exception兜底异常——这意味着即使某个订阅者抛错，也不会中断接收循环。

在 tests/test_event.py 中可以看到Event的完整行为验证，包括同步/异步处理器、异常隔离、emitted等待等场景，说明该机制是框架级、经过测试保证的基础设施。

4.4 用 ui.line_plot 实时推送曲线

@ui.page('/') def page(): line_plot = ui.line_plot(n=1, limit=100, figsize=(10, 4)) number_received.subscribe(lambda number: line_plot.push([datetime.now()], [[number]]))

ui.line_plot是 NiceGUI 基于 matplotlib 封装的实时折线图（源码见 nicegui/elements/line_plot.py）。示例用到的参数：

• n=1：绘制 1 条曲线；
• limit=100：每条线最多保留 100 个数据点，新点到来会顶掉最旧的点（源码中通过self.slice = slice(-limit, None)实现滚动窗口）；
• figsize=(10, 4)：matplotlib 的图形尺寸参数，通过关键字参数透传给pyplot.figure。

push(x, Y)方法接受 x 值列表与 Y 值列表的列表（每个内层列表对应一条线），源码会先追加新数据、再按limit裁剪历史，默认还会自动调整坐标轴范围以贴合最新数据（x_limits/y_limits参数默认为'auto'），最后把 matplotlib 图形重新转换为前端 HTML 渲染。因此回调里每次push([datetime.now()], [[number]])就是在 x 轴上追加当前时间戳、在 y 轴上追加刚收到的数值，形成随时间滚动的实时曲线。

五、运行步骤与预期效果

本示例由两个组件组成：发布者与 NiceGUI 服务器。

5.1 后台运行发布者

python zmq-server.py &

5.2 启动 NiceGUI 服务器

python main.py

默认情况下 NiceGUI 会监听0.0.0.0:8080（可在 nicegui/ui_run.py 的ui.run中查看 host/port 默认值），浏览器访问http://127.0.0.1:8080即可看到页面。

5.3 预期效果

当发布者与 GUI 服务器同时运行后，页面会出现一张随时间实时更新的折线图：随机数（0～100）以每秒约 10 条的速率到达并被绘制，曲线呈现高频上下波动：

5.4 清理后台进程

main.py结束运行后，用如下命令杀掉后台的zmq-server.py（%%在 bash 中代表最近一个后台作业）：

kill -9 %%

六、扩展方向

掌握本示例的骨架后，可以沿以下方向做实战改造：

1. 更换消息内容：将随机整数替换为传感器读数、行情价格或日志指标，消息体用 JSON 编码，接收端json.loads解析后emit结构化数据；
2. 多曲线展示：ui.line_plot(n=2, ...)并订阅不同数据通道，或在回调中按需传入多个 Y 序列；
3. 数据持久化：在read_loop中把收到的数据同时写入 SQLite/Redis（参考仓库内 sqlite_database 与 redis_storage 示例），让看板同时具备历史回放能力；
4. 多消费者扩展：PUSH/PULL 本身支持负载均衡分发，可启动多个 PULL 进程并行消费；
5. 跨进程部署：将tcp://localhost:5555替换为实际主机地址，即可实现远程数据源接入。

七、小结

本示例以约 40 行代码演示了一条完整的"数据生产 → 消息队列传输 → 异步消费 → 前端实时渲染"流水线。其工程价值在于三层解耦：进程级解耦（发布者与 Web 服务器相互独立）、I/O 级解耦（zmq.asyncio让 ZeroMQ 融入 asyncio 事件循环）、逻辑级解耦（Event让后台数据流与 UI 更新互不干扰）。这套模式可以直接复用到任何需要实时可视化的场景，是理解 NiceGUI 异步编程模型的极佳起点。

【免费下载链接】niceguiCreate web-based user interfaces with Python. The nice way.项目地址: https://gitcode.com/GitHub_Trending/ni/nicegui