Flask+ECharts实时生产看板：从轮询到SSE的避坑指南

简介这份源码资源面向从事数据可视化大屏开发的前后端工程师与数据分析人员聚焦制造业生产管理场景提供基于 Echarts 与 Python Flask 的动态实时看板实现方案可用于工厂生产数据监控、订单与出入库信息展示等实际项目参考。压缩包共 142 个文件约 9.96MB以 60 个 js 脚本、20 个 css 样式、9 个 html 页面和 9 个 json 配置为主另含 8 张 png 图片、3 个 py 后端文件及说明文档前端图表渲染与后端数据接口分工清晰便于二次开发与模块替换。目前已有 2611 人学习下载说明该方案在同类大屏项目中具备一定参考价值。读者可从中获取完整的 Flask 服务端结构、Echarts 图表配置与实时刷新逻辑以及制造业看板的页面布局与样式组织方式适合作为课程设计、毕业设计或企业原型开发的起步模板帮助快速搭建可运行的大屏框架并理解前后端数据联动流程。1. 制造业生产管理看板为什么“实时”两个字让一半的 Flask 项目翻车制造业车间里最不缺的就是数据缺的是把数据在正确的时间送到正确的人眼前。很多团队做生产管理看板第一版用 ECharts 画静态图领导看着挺满意等到要接 PLC、MES、扫码枪的实时数据时问题全冒出来了——页面卡死、数据延迟十几秒、刷新一次整块屏闪一下。基于 ECharts Python Flask 的动态实时大屏核心要解决的就是“数据从产线到浏览器这一路怎么走得不堵、不丢、不闪”。这套方案适合两类人一类是工厂信息化岗手头有产线数据但不知道怎么快速搭一个能看的看板另一类是后端或数据开发想用 Python 生态把实时可视化这条链路跑通。它不追求炫酷的 3D 效果追求的是产量、良率、设备状态这些指标能秒级更新而且部署成本可控。下面按“先跑通最小链路再调实时参数最后避坑”的顺序拆开讲。2. 用 Flask ECharts 跑通第一个动态看板从假数据到真接口2.1 为什么选 Flask 而不是 Django 或 FastAPI做实时看板后端框架的选择标准其实就三条启动快、改起来快、和前端对接简单。Flask 在这三点上都很顺手。Django 自带 ORM 和 Admin做看板这种轻量级数据推送场景反而显得重改一个接口要动 settings、urls、views 好几个文件FastAPI 异步性能好但如果团队里有人不熟悉 async/await排查问题会多一层心智负担。Flask 的同步视图在每秒几十次请求的量级下完全够用而且和 ECharts 的 AJAX 轮询配合起来逻辑最直白。我一般会这样搭最小骨架一个app.py提供数据接口一个templates/index.html放 ECharts 容器一个static/js/dashboard.js写轮询和图表更新逻辑。数据源先用 Python 字典模拟等链路通了再把字典换成数据库查询或消息队列消费。2.2 最小可运行代码Flask 接口 ECharts 轮询先看后端接口。下面这段代码提供一个返回 JSON 的/api/production路由模拟产量和良率数据# app.py from flask import Flask, jsonify, render_template import random import time app Flask(__name__) # 模拟产线数据源实际项目中替换为数据库查询或 MQ 消费 def fetch_production_data(): return { timestamp: int(time.time() * 1000), output: random.randint(800, 1200), # 当前班次产量 yield_rate: round(random.uniform(92.0, 99.5), 2), # 良率百分比 device_status: random.choice([运行, 待机, 故障]), stations: [ {name: 工位A, value: random.randint(100, 300)}, {name: 工位B, value: random.randint(100, 300)}, {name: 工位C, value: random.randint(100, 300)}, {name: 工位D, value: random.randint(100, 300)}, ] } app.route(/) def index(): return render_template(index.html) app.route(/api/production) def production(): return jsonify(fetch_production_data()) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)逻辑说明fetch_production_data是数据聚合层真实项目里这里会换成 SQL 查询或从 Redis 读缓存。/api/production每次请求都返回最新快照前端拿到后局部更新图表。debugTrue只在开发时开生产环境必须关掉否则会有性能和安全问题。参数说明host0.0.0.0让局域网内其他机器能访问方便车间大屏直接连port5000是 Flask 默认端口如果和现场其他服务冲突就改成 8080 或 9000。timestamp用毫秒级时间戳前端可以用来判断数据是否过期。前端部分ECharts 的初始化和平滑更新是关键// static/js/dashboard.js var outputChart echarts.init(document.getElementById(outputChart)); var stationChart echarts.init(document.getElementById(stationChart)); // 初始配置只写一次 outputChart.setOption({ title: { text: 实时产量 }, series: [{ type: gauge, data: [{ value: 0, name: 件/班 }] }] }); stationChart.setOption({ title: { text: 工位产量分布 }, xAxis: { type: category, data: [] }, yAxis: { type: value }, series: [{ type: bar, data: [] }] }); function refreshData() { fetch(/api/production) .then(function(res) { return res.json(); }) .then(function(data) { // 只更新数据不重建图表避免闪烁 outputChart.setOption({ series: [{ data: [{ value: data.output, name: 件/班 }] }] }); stationChart.setOption({ xAxis: { data: data.stations.map(function(s) { return s.name; }) }, series: [{ data: data.stations.map(function(s) { return s.value; }) }] }); }) .catch(function(err) { console.error(数据拉取失败, err); }); } setInterval(refreshData, 3000); // 3 秒轮询一次 refreshData();逻辑说明setOption在 ECharts 里是合并更新第二次调用只传变化的部分图表不会整体重绘这是避免闪烁的核心。setInterval的间隔要和后端数据更新频率匹配太短浪费请求太长看着卡顿。参数说明3000毫秒是制造业看板的常用值产量类指标 3 到 5 秒更新一次足够如果是设备报警类建议压到 1 秒但后端要换成 WebSocket 或 SSE轮询扛不住。2.3 把假数据换成真实数据源的三种接法模拟数据跑通后接真实数据常见三条路。第一条是直连数据库用pymysql或sqlalchemy在fetch_production_data里查最新记录适合数据已经在 MySQL 或 PostgreSQL 里的场景缺点是每次请求都查库并发高了数据库压力大。第二条是读 Redis 缓存产线侧写一个采集程序把数据推到 RedisFlask 只负责读响应能压到毫秒级这是我最推荐的方案。第三条是消费消息队列用kafka-python或pika从 Kafka/RabbitMQ 拉数据适合多产线、多车间汇总的场景但架构复杂度也最高。选哪条取决于现有基础设施。如果工厂已经有 MES 数据库直连最快出效果如果从零开始直接上 Redis 缓存层后面扩展不后悔。3. 实时刷新参数怎么调轮询、SSE 还是 WebSocket3.1 三种推送方式的延迟和资源消耗对比看板的“实时”程度直接取决于数据推送方式。轮询最简单前端定时发请求后端返回最新数据延迟等于轮询间隔资源消耗随客户端数量线性增长。SSEServer-Sent Events是服务器单向推浏览器用EventSource接收延迟能到毫秒级但每个连接占一个 Flask 线程客户端多了要上 gevent 或 gunicorn 多 worker。WebSocket 是全双工适合需要前端回传指令的场景但 Flask 原生不支持要加flask-socketio部署时还要处理负载均衡的粘性会话。方式典型延迟服务端压力前端复杂度适用场景轮询1-5 秒中低指标看板、日报表SSE100-500 毫秒中高中实时监控、报警推送WebSocket50-200 毫秒高高双向控制、协同操作我一般会这样选产量、良率这类秒级指标轮询 3 秒完全够用别过度设计设备报警、安全监控必须 SSE 或 WebSocket晚一秒可能就是事故。3.2 SSE 在 Flask 里的最小实现和线程注意事项SSE 的 Flask 实现核心是返回一个生成器用text/event-stream作为 content type# app.py 追加 SSE 路由 from flask import Response import json def event_stream(): while True: data fetch_production_data() # SSE 格式要求data: 开头两个换行结尾 yield fdata: {json.dumps(data, ensure_asciiFalse)}\n\n time.sleep(1) # 服务端控制推送频率 app.route(/api/stream) def stream(): return Response(event_stream(), mimetypetext/event-stream)逻辑说明event_stream是一个无限生成器每次 yield 一条 SSE 消息。ensure_asciiFalse保证中文设备名正常显示。time.sleep(1)控制推送间隔放在服务端比前端轮询更可控。参数说明SSE 默认没有超时但 Nginx 反代会断掉长时间空闲的连接需要在 Nginx 配置里加proxy_read_timeout 3600s和proxy_buffering off。Flask 开发服务器是单线程的SSE 会阻塞其他请求生产环境必须用 gunicorn 加--worker-class gevent或--threads参数。前端接收 SSE 比轮询还简单var source new EventSource(/api/stream); source.onmessage function(event) { var data JSON.parse(event.data); // 更新图表逻辑同上 outputChart.setOption({ series: [{ data: [{ value: data.output, name: 件/班 }] }] }); }; source.onerror function() { console.error(SSE 连接断开浏览器会自动重连); };注意EventSource断线后浏览器默认 3 秒重连不需要自己写重连逻辑。但如果服务端返回了非 200 状态码重连会停止所以 Flask 端要保证 SSE 路由始终返回 200。3.3 轮询间隔和 ECharts 动画时长的配合技巧即使选了轮询也有个容易忽略的参数ECharts 的animationDuration。默认值是 1000 毫秒如果轮询间隔是 3 秒图表每次更新会花 1 秒做动画看起来就是“动一下、停两秒、再动一下”观感很差。我一般会把animationDuration调到 300 到 500 毫秒让动画快速完成数据停留时间更长。outputChart.setOption({ animationDuration: 400, animationEasing: cubicOut, series: [{ type: gauge, data: [{ value: 0, name: 件/班 }] }] });另外如果数据变化幅度很小ECharts 的过渡动画反而会让数字看起来在“抖”。这时候可以关掉动画animation: false直接跳变读数更清晰。这个取舍看场景大屏远距离观看时关动画更实用。4. 避坑实时看板从开发到上线的五个血泪教训4.1 页面刷新一次闪一次数据还对不上现象每次轮询回来整个图表重新绘制屏幕明显闪一下而且新旧数据交替时数字会跳变。原因前端每次都用echarts.init重新创建实例或者setOption时传了notMerge: true导致图表完全重建。解决echarts.init只在页面加载时执行一次后续更新只调setOption且不传notMerge。如果图表容器尺寸变了用chart.resize()而不是重建。4.2 Flask 开发服务器一上产线就卡死现象本地测试好好的部署到车间服务器后几个人同时打开看板接口响应从几十毫秒涨到好几秒。原因app.run()启动的是单线程开发服务器同时处理多个请求时会排队。SSE 连接更是会直接占满线程。解决生产环境用 gunicorn 部署命令是gunicorn -w 4 -k gevent -b 0.0.0.0:5000 app:app。-w 4是 4 个 worker 进程-k gevent用协程处理并发连接。如果用了 SSEworker 数要按“同时在线看板数 / 每个 worker 能扛的连接数”来估算一般 4 到 8 个 worker 够一个车间用。4.3 时间戳对不上前端显示的数据是旧的现象看板上显示的时间和实际时间差了几分钟或者数据明显是上一班的。原因后端返回的时间戳单位不统一有的用秒有的用毫秒或者前端拿本地时间做判断但车间大屏的系统时间没同步。解决统一用毫秒级时间戳前端用new Date(timestamp)转换。大屏机器上装 NTP 客户端和服务器对时或者后端在返回数据时附带服务器时间前端以服务器时间为准。4.4 中文乱码设备名显示成问号现象接口返回的 JSON 里中文正常但前端显示乱码。原因Flask 的jsonify默认ensure_asciiTrue会把中文转成 Unicode 转义序列。虽然浏览器能解析但如果中间有代理或日志系统处理不当就会出问题。解决在 Flask 配置里加app.config[JSON_AS_ASCII] False或者手动用json.dumps(data, ensure_asciiFalse)再包Response。SSE 场景必须手动处理因为jsonify不适用于流式响应。4.5 看板开着没人看服务器资源白耗现象看板 24 小时开着但夜班没人盯服务器一直在跑轮询和 SSE 推送。原因没有做页面可见性检测浏览器切到后台标签页或大屏进入屏保后请求照发。解决用document.visibilityState判断页面是否可见不可见时暂停轮询或断开 SSE可见时恢复。var timer null; function startPolling() { if (timer) return; timer setInterval(refreshData, 3000); } function stopPolling() { clearInterval(timer); timer null; } document.addEventListener(visibilitychange, function() { if (document.visibilityState visible) { startPolling(); } else { stopPolling(); } }); startPolling();这个改动能让服务器在非工作时间的负载降一半以上尤其是多块大屏的场景。5. 让看板真正好用数据聚合层和降级策略5.1 在 Flask 和数据库之间加一层聚合缓存直连数据库的看板一旦产线数据量上来查询会越来越慢。我习惯在 Flask 和数据库之间加一个聚合层用一个后台线程每隔固定时间把原始数据算成看板需要的指标写进 Redis 或内存字典Flask 只读聚合结果。这样接口响应稳定在毫秒级数据库压力也小。# aggregator.py import threading import time import redis r redis.Redis(hostlocalhost, port6379, db0) def aggregate_loop(): while True: # 这里替换为真实的聚合查询 data fetch_production_data() r.set(dashboard:production, json.dumps(data, ensure_asciiFalse)) time.sleep(2) # 聚合频率比前端轮询略快 def start_aggregator(): t threading.Thread(targetaggregate_loop, daemonTrue) t.start()在app.py里启动聚合线程接口改成从 Redis 读from aggregator import start_aggregator, r start_aggregator() app.route(/api/production) def production(): cached r.get(dashboard:production) if cached: return Response(cached, mimetypeapplication/json) return jsonify(fetch_production_data()) # 降级缓存没命中时直查逻辑说明聚合线程独立于 Web 请求运行daemonTrue保证主进程退出时线程自动结束。接口优先读缓存缓存失效时降级直查保证看板不会白屏。参数说明time.sleep(2)是聚合频率比前端 3 秒轮询快一点保证前端每次拿到的都是新鲜数据。Redis 的db0是默认库如果和其他业务共用 Redis建议单独开一个库或加 key 前缀。5.2 接口挂了怎么办前端降级和占位显示实时看板最怕接口挂掉后页面一片空白操作员不知道是没数据还是系统坏了。我一般会在前端做两层降级第一层是请求失败时保留上一次的数据并在角落显示“数据延迟”提示第二层是连续失败超过阈值后切换到静态占位图避免大屏上出现报错信息。var lastData null; var failCount 0; function refreshData() { fetch(/api/production) .then(function(res) { return res.json(); }) .then(function(data) { lastData data; failCount 0; updateCharts(data); hideWarning(); }) .catch(function() { failCount; if (lastData) { updateCharts(lastData); // 保留旧数据 } if (failCount 3) { showWarning(数据连接异常请检查采集服务); } }); }这个策略的核心是“宁可显示旧数据也不显示空白”。制造业现场操作员看到旧数据会知道系统有问题看到空白只会以为看板坏了。5.3 大屏适配分辨率、字体和颜色对比度车间大屏通常是 1920x1080 或 4K观看距离 3 到 5 米。ECharts 的默认字体大小在远距离下偏小我一般会把全局字号调到 16 到 20 像素关键指标产量、良率用 36 像素以上。颜色方面深色背景配高饱和度的绿、黄、红比浅色背景更醒目而且车间灯光复杂时深色底不容易反光。var option { textStyle: { fontSize: 18, color: #e0e0e0 }, backgroundColor: #1a1a2e, title: { textStyle: { fontSize: 24, color: #ffffff } }, series: [{ type: gauge, axisLabel: { fontSize: 16 }, detail: { fontSize: 36, color: #00ff88 } }] };另外大屏浏览器建议全屏模式运行隐藏地址栏和滚动条用window.open或 kiosk 模式启动。如果大屏支持直接设成开机自启浏览器全屏打开看板地址减少人为操作。这套方案我从模拟数据跑到车间实际部署最大的体会是实时看板的难点不在 ECharts 画得多好看而在数据链路稳不稳、异常时有没有兜底。先把轮询跑通再按需升级 SSE最后把聚合缓存和降级策略补上基本就能扛住一个车间的日常使用。希望帮到你。本文还有配套的精品资源点击获取