从智能看板到实时数据链路：MQTT+WebSocket 构建产线监控大屏-北京尧图网络

简介面向企业信息化与数据可视化场景的智能看板前端资源包适合需要快速搭建业务数据展示原型的前端开发者、产品经理或运维人员。压缩包内共收纳十四个文件整体体积仅九百一十四KB以六个JavaScript脚本为核心涵盖jQuery、ECharts及中国地图扩展等常用库搭配一个HTML入口页面、一个CSS样式表以及六张用于界面装饰的PNG或JPG图片素材目录结构清晰便于按需拆分复用。目前已有100人学习下载。资源包含可直接运行的页面骨架与图表配置示例以及折线图、柱状图、地图等可视化组件随附的边框、背景及logo等图片素材可辅助完善看板视觉风格节省从零搭建和检索图表库的时间。整体而言这一轻量级资源包适合用于企业内部管理看板的快速原型验证与二次开发。1. 智能看板为什么产线大屏总在“翻车”提到智能看板很多人第一反应是画一张酷炫的大屏界面。真正落到车间里一个看板最容易翻车的地方不是 UI而是“数据能不能持续、真实地流动”。我拆过一套面向产线的智能看板资源它把设备状态采集、消息队列、WebSocket 推送和前端图表封装成了一个能完整跑起来的工程而不是一张静态设计稿。这套东西适合两类人一类是刚开始做设备监控、需要一套可复现的实时链路做参考的开发者另一类是手里有设备接入需求、想先看一个能落地运行的 demo 再接真实硬件的实施工程师。它解决的痛点是“大屏看了三分钟就不动了”“设备在跑但数值对不上”这些现场问题。如果你的诉求只是做一张周期性更新的报表大屏用不上它如果你追求的是秒级刷新、断线自愈、数据可回放这套思路值得从头到尾走一遍。2. 技术栈与项目骨架先把数据链路建成一条河看板类项目最忌讳上来就写页面。我在拆这个压缩包的时候先做的不是打开前端工程而是把它的数据流向画出来。只有链路通了后面改 UI、加指标才不会伤筋动骨。2.1 看板的技术分层从设备到浏览器要过五层这套智能看板把数据从车间拉到浏览器中间分了五个职责明确的层。每一层都可以单独替换这也是它适合做参考原型的原因。我整理了一张表对应压缩包里的目录结构层级职责本资源里的实现对应目录设备层产生状态、产量、能耗等原始数据用模拟数据生成器替代真实 PLCscripts/simulator接入层采集设备数据并转换为统一格式MQTT 客户端上报 JSONservices/collector消息层缓冲并分发实时消息开源 MQTT 服务端broker 配置目录服务层处理订阅、存历史、推前端FastAPI WebSocketservices/server展示层大屏渲染与交互Vue3 ECharts 页面web 目录值得说明的是“消息层”的设计。很多第一次接触看板的人会把数据库当成中转站让前端隔几秒轮询一次接口来刷数据数据量小的时候没问题设备数量一多就容易把数据库和带宽打满。MQTT 在这套资源里承担的是“河流”的角色设备数据持续地流进来后端服务只在河边取水前端通过 WebSocket 从服务端拿最新的一瓢而不是反复去数据库里舀。2.2 为什么选 MQTT 而不是 HTTP 轮询现场组网和弱网场景下的取舍我在看板选型的时候被问得最多的就是“为什么不用 HTTP 轮询”。这里有个工程直觉HTTP 轮询是“客户端主动问”服务端被动应答轮询间隔设短了线路压力大设备网关也容易被高频请求拖垮设长了大屏上的数据看起来就像动画片掉帧。MQTT 是发布订阅模型设备侧主动上报服务端只要保持一条长连接就能持续收数据体量增大时也只是增加主题和订阅关系不会让数据库多扛查询压力。这套资源里还做了一个很细节的选择设备状态走 QoS 1历史归档走 QoS 0。设备状态丢失一次会影响当前显示但下一秒还会再报历史归档追求吞吐容忍极少量丢失。这个区分在真实产线里很常见——报警和状态数据不能丢普通能耗数据丢了也就丢了。2.3 数据消息的格式设计Topic 和 Payload 定得越规范后面越省心消息格式是看板工程容易被忽略、却又直接决定“数值对不上”的关键。压缩包里把 MQTT 主题设计成三层结构iot/device/{device_id}/metrics iot/device/{device_id}/status主题按设备和数据类型拆开订阅方可以灵活地只关心某台设备的报警状态而不必接收全部消息。这样当车间里有 30 台设备时前端和服务端都不用做大量无谓过滤。每条消息的 Payload 也保持了统一结构下面是一个模拟生成的 JSON 示例{ device_id: CNC-01, ts: 1705392000000, metrics: { oee: 0.86, energy_kwh: 12.7, temperature_c: 46.2 }, status: running }字段device_id用于标识设备ts是毫秒时间戳metrics携带业务指标。设计上刻意把“运行状态”和“计量指标”分到两个字段避免前端为了判断一台设备是否在跑而解析整个指标对象。后来我在接入真实设备时只需要把协议层的数据映射成这个结构后续链路完全不用改。3. 核心功能拆解从设备数据到页面显示的一次心跳这一章要把资源里最核心的三段代码逻辑讲透模拟数据怎么产生、后端怎么接收并解析、WebSocket 怎么把数据推到前端。这些都是能直接复制到项目里改着用的。3.1 模拟数据生成器没设备的时候也能把链路跑通压缩包里自带了一个data_simulator.py它模拟三台设备轮流上报状态。没有真实 PLC 的时候这个脚本就是在替设备层“说话”。import json import random import time import paho.mqtt.client as mqtt # 模拟三台设备设备 ID 分别为 CNC-01、CNC-02、CNC-03 DEVICES [CNC-01, CNC-02, CNC-03] BROKER_HOST 127.0.0.1 BROKER_PORT 1883 SEND_INTERVAL_SEC 2 # 每台设备每 2 秒上报一次 def build_payload(device_id: str) - dict: # 模拟一台加工中心的状态OEE 在 0.6~0.95 之间波动 return { device_id: device_id, ts: int(time.time() * 1000), metrics: { oee: round(random.uniform(0.6, 0.95), 2), energy_kwh: round(random.uniform(8.0, 15.0), 1), temperature_c: round(random.uniform(38.0, 55.0), 1) }, status: running } def on_connect(client, userdata, flags, rc): print(模拟器已连接 MQTT Broker) client mqtt.Client() client.on_connect on_connect client.connect(BROKER_HOST, BROKER_PORT, keepalive30) client.loop_start() while True: for dev in DEVICES: payload build_payload(dev) # 状态主题走 QoS 0让高频上报尽量轻量 client.publish(fiot/device/{dev}/metrics, json.dumps(payload), qos0) time.sleep(SEND_INTERVAL_SEC)这段代码有两个可以实际借鉴的参数。keepalive30表示客户端和 Broker 之间每 30 秒确认一次心跳如果中间网络断开Broker 能在约 45 秒内感知并把离线状态推给订阅方SEND_INTERVAL_SEC控制上报频率真实设备如果来自数据采集网关一般也会给成一个可配置值而不是写死在代码里。我通常会把这两个值提到配置文件里方便在弱网环境下把保持时间调大同时降低上报频率。3.2 后端订阅与解析把 MQTT 消息变成结构化数据后端的collector进程负责订阅所有设备的指标主题并把数据写入本地数据库。这段代码的核心是on_message回调它把来自 MQTT 的字节流解析成 Python 字典。import json import sqlite3 import paho.mqtt.client as mqtt from datetime import datetime BROKER_HOST 127.0.0.1 DB_PATH ./data/board.db def insert_metric(conn, device_id: str, ts: int, metrics: dict): # 把每台设备的最新指标存入独立表供后续回放与排查 conn.execute( INSERT INTO metrics(device_id, ts, oee, energy_kwh, temperature) VALUES(?, ?, ?, ?, ?), (device_id, ts, metrics.get(oee), metrics.get(energy_kwh), metrics.get(temperature_c)) ) conn.commit() def on_message(client, userdata, msg): # 只处理指标主题状态主题交给另一条订阅逻辑 try: payload json.loads(msg.payload.decode(utf-8)) dev_id payload[device_id] ts payload[ts] metrics payload[metrics] insert_metric(userdata[conn], dev_id, ts, metrics) except Exception as ex: print(f解析失败: {ex}) conn sqlite3.connect(DB_PATH, check_same_threadFalse) client mqtt.Client(userdata{conn: conn}) client.on_message on_message client.connect(BROKER_HOST, BROKER_PORT, keepalive30) client.subscribe(iot/device/#, qos1) client.loop_forever()这里有个容易踩坑的细节sqlite3.connect默认不允许跨线程使用连接而后端既要跑 MQTT 循环又要处理 WebSocket 推流所以我传入了check_same_threadFalse。如果你的并发量上来还是建议换成真正的连接池或者独立的消息队列把入库操作和推流操作拆到不同的进程里去。3.3 WebSocket 推送前端不是去“查”数据而是“等”数据后端收到数据后需要把变化及时通知浏览器。这里用的是 FastAPI 的 WebSocket 端点后端维护一组活跃连接有新数据进来就广播给所有前端页面。from fastapi import FastAPI, WebSocket from typing import List import asyncio app FastAPI() active_connections: List[WebSocket] [] app.websocket(/ws/board) async def board_socket(ws: WebSocket): await ws.accept() active_connections.append(ws) try: while True: # 等待前端发送心跳维持连接不被代理层断开 data await ws.receive_text() if data ping: await ws.send_text(pong) except Exception: pass finally: active_connections.remove(ws) async def broadcast_metric(payload: dict): # 序列化一次复用给所有连接降低重复编码开销 text json.dumps(payload, ensure_asciiFalse) for conn in active_connections[:]: try: await conn.send_text(text) except Exception: active_connections.remove(conn) async def push_loop(): while True: row fetch_latest_metric_from_db() if row: payload { device_id: row[0], ts: row[1], oee: row[2], energy_kwh: row[3] } await broadcast_metric(payload) await asyncio.sleep(1)broadcast_metric里做了一个很小的优化先把字典序列化成字符串再逐连接发送。这样可以避免十台大屏连接都各自执行一次json.dumps。另外前端会定期发送ping这不仅是保活还能主动发现失效的连接。很多看板页面长时间挂着后数据不再更新就是因为没有做心跳检测连接已经断了应用还蒙在鼓里。4. 本地复现从零把看板跑起来的三步法下载解压之后很多人急着打开前端页面结果自然是白屏。复现这套东西的正确步骤是“先起消息层再起服务层最后开页面”。下面是我在干净环境里完整跑通的过程。4.1 环境准备Python 依赖与目录检查我一般会先建一个虚拟环境避免把依赖装到全局。压缩包里附了requirements.txt核心依赖是paho-mqtt、fastapi、uvicorn、sqlite3标准库。python3 -m venv venv source venv/bin/activate pip install -r requirements.txtrequirements.txt里没有锁死所有传递依赖的版本只约束了主要包的版本下限。如果你在复现时遇到某个库不兼容优先看是不是 Python 版本过高导致的而不是急着升级所有包。这套资源按 Python 3.8 到 3.10 测试过更高的版本我也没遇到问题但依赖冲突时有发生。另外解压后请先确认目录结构完整。我每次都会检查有没有web/index.html和services/server/main.py这两个文件缺少任何一个都说明解压不完整后面所有步骤都会白做。4.2 启动消息层和服务层三个进程缺一不可接着启动 MQTT 服务端然后跑两个 Python 进程。为了便于观察日志我习惯把三个终端并排打开。# 终端 1启动 MQTT Broker按资源内配置启动 mosquitto -c ./broker/mosquitto.conf # 终端 2启动后端服务和数据采集 uvicorn services.server.main:app --host 0.0.0.0 --port 8000 python services/collector/main.py # 终端 3启动模拟设备数据源 python scripts/simulator/data_simulator.pymosquitto.conf里配置了端口 1883 和允许匿名访问。如果你在部署到公网服务器一定记得关掉匿名访问并加用户名密码否则任何知道地址的人都能往主题里发消息直接伪造设备数据。我在本地复现时是不加鉴权的省事优先。三个进程都起来后看板首页数据会开始滚动更新。如果页面显示但数据不动优先去终端 3 看模拟器有没有正常打印“已连接”日志再检查 MQTT 端口是否被防火墙拦截。排查顺序永远是“数据源 → 消息层 → 消费端”不要一上来就改前端代码。4.3 验证复现成功三个指标确认链路健康页面能动不代表链路是通的。我有一套快速验证的检查点能在五分钟内判断整个看板是否健康。检查点方法预期结果数据源上报在终端 3 观察日志每 2 秒有消息打印消息层接收用命令行订阅端查看主题iot/device/CNC-01/metrics有消息流入页面刷新打开浏览器开发者工具的“网络”面板WebSocket 帧可见每 1~2 秒有数据如果前两个检查通过但页面没有更新问题基本出现在 WebSocket 端到端的推送逻辑上重点看后端有没有抛异常。这个方法我后来在接真实项目时也在用确认链路比美化界面重要得多。5. 避坑排查看板数据不来、图表不动的五个元凶下面这几条是我复现这套资源以及基于它改造真实项目时踩过的坑。每一条都按“现象、原因、解决”来写照着排查能省下不少时间。5.1 图表一直不刷新浏览器控制台提示连接失败现象前端正常打开但图表上的数字永远停在一个值上控制台提示无法连接 WebSocket。原因绝大多数情况不是代码问题而是跨域或端口访问限制。看板前端和后端如果部署在不同端口后端 WebSocket 没有明确允许跨域来源浏览器会把连接请求拦掉。解决在后端加允许跨域的中间件并确认前端ws://地址里的端口号与实际后端端口一致。如果大屏部署在其他机器上还要检查服务器的防火墙有没有放行对应的 TCP 端口。我见过不少项目因为“本机能跑、局域网设备连不上”而卡住最后发现只是防火墙没放开。5.2 数据延迟越来越大页面从秒级变成分钟级现象一开始数据刷新很快跑半小时后页面数据越来越旧刷新浏览器才恢复。原因消息积压。模拟器上报频率在产线场景下可能被调高而订阅端入库和推送是串行的处理不过来就会在回调里堆积。另外如果订阅时用了 QoS 2Broker 会额外做多次确认设备多时也会显著增加延迟。解决把高频指标改为 QoS 0低频报警保留 QoS 1同时把入库操作改为独立线程或批量写入。后端进程内部用队列接收解析结果消费端单独提交数据库事务相当于把一条单行道改成两条并行通道。5.3 运行几天后内存一路往上飙页面越来越卡现象看板刚启动时内存 200MB三五天后涨到 2GB甚至被系统杀掉。原因前端在反复创建图表实例后没有销毁旧实例后端侧则是长期运行积累了大量的 SQLite 连接和未清理的临时对象。ECharts 图表在数据更新时如果多次执行init而不是setOption每一个旧实例都会驻留在内存里。解决前端统一用setOption更新图表销毁页面时调用dispose()后端定期清理无效的 WebSocket 连接同时给数据库写入加上批量提交策略避免每个指标都开一次写事务。5.4 页面时间显示成 NaN或比实际时间差 8 小时现象图表横轴时间全是NaN或者显示的时间比车间时钟早了整整八个小时。原因时间戳单位不统一。后端传给前端的是毫秒前端却按照秒去解析如果后端存的是 UTC 时间前端直接取本地时区渲染也会出现 8 小时偏移。解决在数据接入层统一把时间戳转换成毫秒并且前后端约定好“时间戳一律是 UTC 毫秒”前端渲染时再转成本地时间。我习惯在模拟器里生成时间戳时就明确用毫秒同时在字段注释里写清楚单位避免后续接手的同事猜谜。5.5 看板在大屏主机上播放时 CPU 占用高、画面掉帧现象同一个页面在开发电脑上很流畅放到工控主机大屏上后风扇狂转、画面明显卡顿。原因工控主机显卡性能有限页面却做了大量持续动画和阴影特效浏览器没有开启硬件加速时CPU 要承担所有绘制计算。解决把页面中不必要的动画关闭例如数值闪烁、平滑过渡效果在浏览器设置里开启硬件加速。如果还卡降低刷新频率从每秒推送改为每两秒推送一次。大屏上“稳定展示”比“丝滑动画”更重要这是现场运营人员最在意的一点。6. 从单机看板到多场景复用把灵活性留在后端配置里看完板能不能从一个产品变成一套可复用的工程我的判断标准是换一个数据源、加一块新看板需要改动多少处代码。这套资源里预留了几个关键的扩展点顺着这些点改造单机 Demo 也能变成多产线共用的平台底座。6.1 数据源从模拟器切换到真实采集网关资源里的模拟器和真实接入最大的区别只在于数据入口。真实场景下采集网关通常已经做好了 Modbus TCP 或 OPC UA 到 MQTT 的转换后端要改的只是主题前缀和字段映射。比如某台真实设备的主题是factory_a/plant_2/machine_07/status后端只需要在订阅配置里增加一个映射表订阅主题映射设备 ID额外处理factory_a/plant_2/machine_07/statusFA-P2-M07无factory_a/plant_2/machine_09/statusFA-P2-M09温度超限时触发告警这个映射我一般放在config.yaml里后端的订阅逻辑启动时读取配置动态生成不需要改main.py。6.2 增加告警同步当指标越界时自动推送看板不应该只显示数据还要能提醒人。我在改造过程中加了一个很实用的功能当温度超过设定阈值时通过 Webhook 把告警消息推到协同工具。实现逻辑很薄只需要在入库前判断一下import requests ALERT_RULES [ {device_id: CNC-01, metric: temperature_c, max: 50.0} ] def check_alert(payload: dict): device_id payload[device_id] metrics payload[metrics] for rule in ALERT_RULES: if rule[device_id] device_id: value metrics.get(rule[metric]) limit rule[max] if value is not None and value limit: msg f设备 {device_id} 温度 {value} 超过阈值 {limit} requests.post(http://your-webhook-url, json{text: msg}) return阈值规则放在列表里可以随意增删不需要改动业务代码。这样看板的角色从“给人看”变成了“替人盯”在现场的价值直接提升一个层次。6.3 验证数据一致性让前端数值经得起审计上线一段时间后运营人员会拿看板数值和财务系统的能耗台账核对。为了防止两边对不上我养成了一个习惯每天写一个一致性校验脚本从数据库里导出当天的指标汇总和页面展示的总数做比对。核对脚本通过 SQL 查询数据库当天energy_kwh之和再对比前端展示值差值超过 1% 就视为告警。从那以后我每次接手看板项目都会强制先跑一遍模拟数据链路、确认 WebSocket 帧正常、再做一次数值校验三个步骤全过才开始画界面。这个习惯让我少背了很多“看板数据不准”的锅。希望帮到你。本文还有配套的精品资源点击获取

从智能看板到实时数据链路：MQTT+WebSocket 构建产线监控大屏

相关推荐

WPF窗体样式重写实战：ControlTemplate与WindowChrome深度解析

Linux系统CPU使用率低但负载高？四类原因与排查指南

libIEC61850 pyiec61850 Python 绑定：从 CMake 构建到 RCB 订阅回调的完整实践指南

更多资讯

手写可定制搜索框：从输入层到渲染层的完整实现

liteLLM 实战：搭建统一大模型网关，搞定多模型路由与故障转移

7种高级RAG架构实战指南：从混合检索到知识图谱增强

HTML表单完全指南：从标签属性到提交原理与实战避坑

PHP实现银行卡号识别：BIN号匹配原理与Redis缓存实战

t3code 代码转换流水线：IR 设计、规则引擎与渲染机制实战

今日更新

声子晶体仿真：三角晶格与六角晶格的能带计算实战

度假村安保定岗定编：岗位梳理、编制测算与排班设计

COMSOL等离子体模块实战：氩气双层DBD仿真建模全解析

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？