使用 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 的邂逅本示例构建了一套极简的实时数据流水线包含两个独立进程发布者publisherzmq-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 明确给出了版本下限nicegui3.0 pyzmq 26.0pyzmq即 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(fSending 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(n1, limit100, 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(fReceived 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:5555zmq.asyncio.Poller提供事件驱动的多路复用await poller.poll()会挂起当前协程直到有注册的套接字可读从而避免忙轮询浪费 CPUzmq.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(fReceived 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.pysubscribe还做了内存安全兜底当在 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(n1, limit100, figsize(10, 4)) number_received.subscribe(lambda number: line_plot.push([datetime.now()], [[number]]))ui.line_plot是 NiceGUI 基于 matplotlib 封装的实时折线图源码见 nicegui/elements/line_plot.py。示例用到的参数n1绘制 1 条曲线limit100每条线最多保留 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 服务器同时运行后页面会出现一张随时间实时更新的折线图随机数0100以每秒约 10 条的速率到达并被绘制曲线呈现高频上下波动5.4 清理后台进程main.py结束运行后用如下命令杀掉后台的zmq-server.py%%在 bash 中代表最近一个后台作业kill -9 %%六、扩展方向掌握本示例的骨架后可以沿以下方向做实战改造更换消息内容将随机整数替换为传感器读数、行情价格或日志指标消息体用 JSON 编码接收端json.loads解析后emit结构化数据多曲线展示ui.line_plot(n2, ...)并订阅不同数据通道或在回调中按需传入多个 Y 序列数据持久化在read_loop中把收到的数据同时写入 SQLite/Redis参考仓库内 sqlite_database 与 redis_storage 示例让看板同时具备历史回放能力多消费者扩展PUSH/PULL 本身支持负载均衡分发可启动多个 PULL 进程并行消费跨进程部署将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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/14 7:45:05 · 阅读 1.2k 使用 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 的邂逅本示例构建了一套极简的实时数据流水线包含两个独立进程发布者publisherzmq-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 明确给出了版本下限nicegui3.0 pyzmq 26.0pyzmq即 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(fSending 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(n1, limit100, 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(fReceived 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:5555zmq.asyncio.Poller提供事件驱动的多路复用await poller.poll()会挂起当前协程直到有注册的套接字可读从而避免忙轮询浪费 CPUzmq.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(fReceived 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.pysubscribe还做了内存安全兜底当在 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(n1, limit100, figsize(10, 4)) number_received.subscribe(lambda number: line_plot.push([datetime.now()], [[number]]))ui.line_plot是 NiceGUI 基于 matplotlib 封装的实时折线图源码见 nicegui/elements/line_plot.py。示例用到的参数n1绘制 1 条曲线limit100每条线最多保留 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 服务器同时运行后页面会出现一张随时间实时更新的折线图随机数0100以每秒约 10 条的速率到达并被绘制曲线呈现高频上下波动5.4 清理后台进程main.py结束运行后用如下命令杀掉后台的zmq-server.py%%在 bash 中代表最近一个后台作业kill -9 %%六、扩展方向掌握本示例的骨架后可以沿以下方向做实战改造更换消息内容将随机整数替换为传感器读数、行情价格或日志指标消息体用 JSON 编码接收端json.loads解析后emit结构化数据多曲线展示ui.line_plot(n2, ...)并订阅不同数据通道或在回调中按需传入多个 Y 序列数据持久化在read_loop中把收到的数据同时写入 SQLite/Redis参考仓库内 sqlite_database 与 redis_storage 示例让看板同时具备历史回放能力多消费者扩展PUSH/PULL 本身支持负载均衡分发可启动多个 PULL 进程并行消费跨进程部署将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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考