简介:这是基于 Python Flask 框架编写的 SPC 统计过程控制软件工程包,面向质量管理、生产监控场景下的数据分析和开发人员,也适合想学习 Flask 项目实战或控制图算法实现的读者。软件实现了 SPC 控制图核心算法,通过前端页面实时展示数据趋势和异常点,能帮助用户监控生产过程的稳定性,及时发现质量波动。资源包共 2000 个文件,压缩后约 178MB,主要涵盖 Python 源码(py/pyc)、前端 JS/CSS/HTML、数据文件(dat/csv)、图片与日志等类型;py/pyc 对应后端请求处理与算法逻辑,JS/CSS/HTML 构建可视化界面,dat/csv 保存过程数据,png/jpg 为图表或界面资源,Logs、uploads、migrations 等目录则分别承担运行日志、上传数据与数据库版本管理。项目目录结构较完整,适合直接阅读或二次开发。当前已有 378 人浏览学习,对需要搭建质量监控系统、或在 Flask 中集成 SPC 图表的开发者来说,是一份有参考价值的实际工程案例。
1. 自己写 SPC 软件:为什么 Python Flask 是最稳的组合
自己用 Python Flask 写 SPC 软件,最难的从来不是 Flask 配置,而是 SPC 的数据模型和计算逻辑怎么落。SPC(统计过程控制)用控制图实时判断过程是否稳定,避免等不合格品出现再返工。商业套件按点数收费、算法不透明,很多工厂最终选择自研——Python 算统计顺手,Flask 足够轻,不必为一张控制图引入微服务架构。
下文按落地路径走:建数据模型、实现 Xbar-R 控制图和判异规则、用 Flask 路由加 ECharts 出看板,最后是部署和定时重算控制限的坑。
适合想用 Python 替代手工画图的质量工程师,以及要交付工厂自研项目的 Python 开发;整套实现控制限计算核心只有几十行代码。
2. SPC 数据模型与 Flask 项目骨架:从表设计到蓝图拆分
2.1 SPC 实体关系:特性、子组、测量值为什么要分层建模
SPC 软件的建模对象不是“一堆测量值”,而是“一个被监控的特性在什么时间点被抽了多大一组样本”。如果不分层,直接把散点画出来,后面的控制限计算、判异报警、追溯批次全都无从谈起。原因有三个:同一台设备上不同尺寸规格服从不同分布,必须用特性(Feature)隔离;控制图以子组(Subgroup)为分析单元,不是以单件为单元;控制限会随数据滚动更新,需要能回查每次重算的依据。
我见过不少自研 SPC 的常见误区:把每个测量值拆成一行存一张 measurement 表,查询时再按时间窗 group by。这个设计的问题在于分组边界难统一,采样时间抖动几秒钟就会造成误分组,而且均值、极差这类统计函数散落在各处,换个页面就要重写一遍。
正确做法是让子组成为事务单元。一次采样抽 5 件,就在一个事务里写入一个子组记录,原始测量值以 JSON 或分隔字符串冗余存放。这样回查原始数据方便,子组统计量(均值、极差)也随行落库,控制图接口直接读统计列,不用每次现算。
2.2 用 SQLAlchemy 落地四张核心表
下面是我常用的四张表:特性表、子组表、控制限快照表、判异记录表。快照表的作用是让控制限不随每次页面刷新重新计算,判异记录表则是给质量追溯留痕。
# models.py from datetime import datetime from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() class Feature(db.Model): """被监控的计量型特性,每个特性一套规格限和控制图参数""" __tablename__ = 'spc_feature' id = db.Column(db.Integer, primary_key=True) code = db.Column(db.String(32), unique=True, nullable=False, index=True) name = db.Column(db.String(64), nullable=False) usl = db.Column(db.Float) # 规格上限,无上限填 None lsl = db.Column(db.Float) # 规格下限,无下限填 None target = db.Column(db.Float, default=0.0) subgroup_size = db.Column(db.Integer, default=5) # 子组容量 n created_at = db.Column(db.DateTime, default=datetime.utcnow) class Subgroup(db.Model): """一个子组:同一时刻、同一批次抽出的 n 件测量值""" __tablename__ = 'spc_subgroup' __table_args__ = ( db.Index('ix_feature_time', 'feature_id', 'sample_time'), ) id = db.Column(db.Integer, primary_key=True) feature_id = db.Column(db.Integer, db.ForeignKey('spc_feature.id'), nullable=False) batch_no = db.Column(db.String(32), default='') # 批次号,追溯用 sample_time = db.Column(db.DateTime, nullable=False) # 采样时间 values_json = db.Column(db.Text, nullable=False) # 原始测量值,逗号分隔 mean = db.Column(db.Float, nullable=False) # x_bar,落库避免重复计算 rng = db.Column(db.Float, nullable=False) # 极差 R created_at = db.Column(db.DateTime, default=datetime.utcnow) class ControlLimitSnapshot(db.Model): """控制限快照:记录某个特性在某日之后生效的 UCL/CL/LCL""" __tablename__ = 'spc_limit' id = db.Column(db.Integer, primary_key=True) feature_id = db.Column(db.Integer, db.ForeignKey('spc_feature.id'), nullable=False) effective_date = db.Column(db.Date, nullable=False) # 生效日期,按天滚动 x_ucl = db.Column(db.Float) x_cl = db.Column(db.Float) x_lcl = db.Column(db.Float) r_ucl = db.Column(db.Float) r_lcl = db.Column(db.Float) base_points = db.Column(db.Integer, default=25) # 重算用了多少子组 class RuleAlarm(db.Model): """判异记录:规则编号、命中点位、描述,供质量追溯""" __tablename__ = 'spc_rule_alarm' id = db.Column(db.Integer, primary_key=True) subgroup_id = db.Column(db.Integer, db.ForeignKey('spc_subgroup.id'), nullable=False) rule_no = db.Column(db.Integer, nullable=False) description = db.Column(db.String(64)) created_at = db.Column(db.DateTime, default=datetime.utcnow)字段设计的几个取舍:values_json用文本冗余存原始测量值,省掉一张子表,但要由应用层保证长度和格式;mean和rng落库,图表接口和判异算法都直接读这两列,避免重复聚合;复合索引(feature_id, sample_time)是后面查询性能的根基,单列索引在跨特性查询时会失效一半。
2.3 Flask 应用工厂与蓝图注册
PyCharm 新建 Flask 项目时会生成单文件 app.py,开发初期很爽,但控制图计算、录入接口、特性管理全堆在一个文件里,改一处配置就要全局扫雷。我一般一上来就按应用工厂拆,理由有三:测试时能注入不同配置、多环境切换只改配置类、蓝图之间不会互相 import 造成循环依赖。
# app.py from flask import Flask from config import Config from models import db def create_app(config_class=Config): app = Flask(__name__) app.config.from_object(config_class) db.init_app(app) from routes.chart import chart_bp from routes.measure import measure_bp from routes.feature import feature_bp app.register_blueprint(chart_bp, url_prefix='/spc') app.register_blueprint(measure_bp, url_prefix='/api') app.register_blueprint(feature_bp, url_prefix='/feature') with app.app_context(): db.create_all() # 首次启动建表,后续结构变更走 Flask-Migrate return app蓝图按业务域拆:measure负责数据录入,chart负责出图数据,feature负责特性管理。db.create_all()只适合首次建表,等表结构要改的时候必须换成 Alembic 迁移,这个坑在第 5 章会专门讲。如果是内部管理界面不想手写,特性表和子组表的维护页可以直接挂 Flask-Admin 这类后台管理插件,注册一个 ModelView 就有增删改查,省下写表单页面的时间。
2.4 数据库连接配置:SQLite 调试、MySQL 上生产
本地验证算法时用 SQLite 零成本起步,但上头一到多人录入就得切 MySQL。连接配置里最容易忽略的不是地址和账密,而是池化参数。
# config.py import os class Config: SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-only') SQLALCHEMY_DATABASE_URI = os.environ.get( 'DATABASE_URL', 'mysql+pymysql://spc:spc@127.0.0.1:3306/spc_db?charset=utf8mb4' ) SQLALCHEMY_ENGINE_OPTIONS = { 'pool_size': 10, 'pool_recycle': 1800, # 小于 MySQL wait_timeout 默认 8 小时,提前回收连接 'pool_pre_ping': True, # 取连接前先 ping 探活,避免“连接已断开”的幽灵报错 }pool_recycle不设的话,MySQL 空闲 8 小时后主动断连,连接池里的旧连接继续用就会报Lost connection。pool_pre_ping每次取连接前多一次探活,几百条记录的场景下开销可忽略,但能省掉大量半夜报警。首次建库建用户用下面这段 SQL,注意字符集必须指定 utf8mb4,不然中文批次号入库会乱码:
CREATE DATABASE spc_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER 'spc'@'%' IDENTIFIED BY 'spc'; GRANT ALL PRIVILEGES ON spc_db.* TO 'spc'@'%'; FLUSH PRIVILEGES;提示:切换数据源只改
DATABASE_URL环境变量,算法层完全不感知底层是 SQLite 还是 MySQL,这就是把统计逻辑独立成函数、不散落在路由里的价值。
3. Xbar-R 控制图与判异规则:Python 计算核心怎么写
3.1 控制限公式和 A2/D3/D4 系数表
Xbar-R 图由两张图组成:Xbar 图监控子组均值的波动,R 图监控子组极差的波动。控制限公式不复杂,但系数表必须抄对:
Xbar 图: UCL = X̿ + A2·R̄ CL = X̿ LCL = X̿ - A2·R̄ R 图: UCL = D4·R̄ CL = R̄ LCL = D3·R̄ 过程西格玛估计: σ̂ = R̄ / d2其中 X̿ 是所有子组均值的均值,R̄ 是所有子组极差的均值。为什么用极差而不是标准差来估计西格玛:极差对个别异常值更稳健,计算也简单,这是 SPC 在车间场景下选择 R 图的根本原因。子组容量 n 小于 10 用 Xbar-R,n 大于 10 时极差信息损失变大,业界通常换 Xbar-S 图,这个边界要在选型时写进注释。
常用系数表:
| n | A2 | D3 | D4 | d2 |
|---|---|---|---|---|
| 2 | 1.880 | 0 | 3.267 | 1.128 |
| 3 | 1.023 | 0 | 2.574 | 1.693 |
| 4 | 0.729 | 0 | 2.282 | 2.059 |
| 5 | 0.577 | 0 | 2.114 | 2.326 |
| 6 | 0.483 | 0 | 2.004 | 2.534 |
| 7 | 0.419 | 0.076 | 1.924 | 2.704 |
| 8 | 0.373 | 0.136 | 1.864 | 2.847 |
注意 D3 在 n 小于等于 6 时恒为 0,意味着 R 图的下控制限贴零轴,实际只有上侧报警。很多初学者把 D3 填成负数,画出来的 R 图下界是负值,极差不可能为负,这条线毫无意义。
3.2 子组统计与 Xbar-R 控制限计算代码
计算模块独立成spc_core.py,不进蓝图,不依赖 Flask 上下文,这样单测和命令行脚本都能直接调用。
# spc_core.py from statistics import mean A2 = {2: 1.880, 3: 1.023, 4: 0.729, 5: 0.577, 6: 0.483, 7: 0.419, 8: 0.373} D3 = {2: 0.0, 3: 0.0, 4: 0.0, 5: 0.0, 6: 0.0, 7: 0.076, 8: 0.136} D4 = {2: 3.267, 3: 2.574, 4: 2.282, 5: 2.114, 6: 2.004, 7: 1.924, 8: 1.864} D2 = {2: 1.128, 3: 1.693, 4: 2.059, 5: 2.326, 6: 2.534, 7: 2.704, 8: 2.847} def subgroup_stats(values): """计算子组均值与极差。values 是长度 >=2 的数值列表""" if len(values) < 2: raise ValueError('子组至少需要 2 个测量值') x_bar = mean(values) rng = max(values) - min(values) return x_bar, rng def xbar_r_limits(feature, subgroups): """根据历史子组计算控制限。subgroups 按 sample_time 升序传入""" n = feature.subgroup_size if n not in A2: raise ValueError(f'子组容量 n={n} 不在系数表内,请扩展 A2/D3/D4 表') if len(subgroups) < 20: raise ValueError('控制限重算至少需要 20 个子组,点子太少没有统计意义') x_double_bar = mean(s.mean for s in subgroups) r_bar = mean(s.rng for s in subgroups) sigma_hat = r_bar / D2[n] return { 'x_bar': x_double_bar, 'r_bar': r_bar, 'sigma_hat': sigma_hat, # 过程西格玛估计,不四舍五入 'x_ucl': x_double_bar + A2[n] * r_bar, 'x_lcl': x_double_bar - A2[n] * r_bar, 'r_ucl': D4[n] * r_bar, 'r_lcl': D3[n] * r_bar, }逻辑说明:feature只取subgroup_size,控制限完全由历史子组决定。内部计算一律不四舍五入,判异规则要用未舍入的值,舍入只发生在序列化给前端时。少于 20 个子组拒绝重算是一条工程经验,控制图规范要求初始建图至少 25 个子组,取 20 是留了余量的保守值。
3.3 判异规则的 Python 实现
行业里这套规则叫 Nelson 规则或 Western Electric 规则,不同手册对连续点数的定义略有差异。常用的有 8 条,我实现的顺序按出现频率排:
- 单点超出 3σ 控制限
- 连续 9 点落在中心线同一侧
- 连续 6 点单调上升或下降
- 连续 14 点交替升降
- 连续 3 点中有 2 点落在同侧 2σ 之外
# spc_rules.py def detect_rules_xbar(means, center, ucl, lcl): """ 对 Xbar 序列执行判异准则 1~5。 means: 按时间升序的子组均值;center/ucl/lcl 来自 xbar_r_limits。 返回报警列表,point 从 1 开始计数。 """ alarms = [] m = len(means) if m == 0: return alarms zone = (ucl - center) / 3.0 # Xbar 图的 1σ 带宽(子组均值的标准差) z = [(v - center) / zone for v in means] # 规则 1:单点超出 3σ 控制限 for i, v in enumerate(means): if v > ucl or v < lcl: alarms.append({'point': i + 1, 'rule': 1, 'desc': '单点超限'}) # 规则 2:连续 9 点落在中心线同一侧 for i in range(m - 8): seg = means[i:i + 9] if all(v > center for v in seg) or all(v < center for v in seg): alarms.append({'point': i + 9, 'rule': 2, 'desc': '连续9点同侧'}) # 规则 3:连续 6 点单调上升或下降(5 个相邻差同号) for i in range(m - 6): diffs = [means[i + j] - means[i + j - 1] for j in range(1, 6)] if all(d > 0 for d in diffs) or all(d < 0 for d in diffs): alarms.append({'point': i + 6, 'rule': 3, 'desc': '连续6点趋势'}) # 规则 4:连续 14 点交替升降(13 个升降标志两两相反) for i in range(m - 13): seg = means[i:i + 14] flags = [seg[j] > seg[j + 1] for j in range(13)] if all(flags[j] != flags[j + 1] for j in range(12)): alarms.append({'point': i + 14, 'rule': 4, 'desc': '连续14点交替'}) # 规则 5:连续 3 点中有 2 点落在同侧 2σ 之外 for i in range(m - 2): seg = z[i:i + 3] if sum(1 for x in seg if x > 2) >= 2 or sum(1 for x in seg if x < -2) >= 2: alarms.append({'point': i + 3, 'rule': 5, 'desc': '3点中2点超2σ'}) return alarms参数说明:zone的取值是关键细节。Xbar 图的 3σ 控制限对应的是子组均值分布的标准差,等于过程西格玛除以根号 n,所以不能用sigma_hat直接当 1σ 带,而要用(UCL - CCL) / 3反推。规则 4 的实现用相邻“升/降”标志的异或判断,比比较相邻差值的正负号更直观,也避免在零差值的边界上误判。
剩下三条规则:6(5 点中 4 点落在同侧 1σ 之外)、7(连续 15 点落在 1σ 带内)、8(连续 8 点超出 1σ 带两侧)。7 和 8 描述的是数据被人为分层或挑选后的特征,真实产线出现频率低,我通常只把规则 6 补上就收手,规则太多会淹没有效报警。R 图是单侧图,实际只启用规则 1 的超上界判断。
3.4 过程能力 Cp/Cpk 与正态性前提
控制图画完,业务方下一步必然问“过程能力够不够”。Cpk 的计算必须在过程受控(控制图无报警)之后做,否则算出来的能力指数没有任何预测意义。
def cp_cpk(usl, lsl, center, sigma_hat): """规格上下限都定义才计算;返回 (cp, cpk)""" if usl is None or lsl is None: return None, None cp = (usl - lsl) / (6 * sigma_hat) cpu = (usl - center) / (3 * sigma_hat) cpl = (center - lsl) / (3 * sigma_hat) return round(cp, 3), round(min(cpu, cpl), 3)Cpk 取 CPU 和 CPL 的小值,反映过程中心偏离规格中心后的真实能力。判断依据行业内常用:Cpk 小于 1.0 过程能力不足,1.0 到 1.33 勉强,1.33 以上才谈得上充裕。SPC 的统计推断默认测量值近似正态,如果数据明显偏态(比如单向受力的形变数据),先用 Shapiro-Wilk 检验做个快速筛查:
from scipy.stats import shapiro stat, p = shapiro(raw_values) if p < 0.05: # 拒绝正态假设,控制图和 Cpk 结论要打折,考虑 Box-Cox 变换后再分析 pass提示:把 Cp/Cpk 和正态性检验放进同一个模块,是为了让质量工程师在看板上一眼看到“先判稳、再算能力”的完整链路,而不是只给一个数字。
4. Flask 路由与 ECharts 看板:SPC 图表怎么变成 Web 界面
4.1 接口设计:录入、查询、管理三类路由
SPC 软件的 Web 层只需要三类接口:测量值录入、图表数据查询、特性管理。接口路径按功能域划分,返回统一结构,前端不关心后端的计算细节。
| 方法 | 路径 | 作用 | 返回体关键字段 |
|---|---|---|---|
| POST | /api/measurements | 录入一个子组 | x_bar、r、code |
| GET | /spc/chart_data?feature_code=xx | 控制图数据和控制限 | points、limits |
| GET | /spc/alarms?feature_code=xx | 判异报警记录 | alarms |
| POST | /feature | 新建监控特性 | id、code |
chart_data是核心接口,返回 JSON 结构固定,前端直接拿来画图:
{ "feature": "轴承外径", "limits": {"x_ucl": 50.023, "x_cl": 50.001, "x_lcl": 49.979, "r_ucl": 0.045}, "points": [ {"sample_time": "2025-06-01 08:00", "mean": 50.012, "r": 0.021, "alarms": [2, 5]} ] }接口调试时先用 curl 固定住返回结构,再写前端,前后端同时试错很难定位问题。VSCode 里配好 REST Client 插件,或者直接用 Flask 的测试客户端写接口测试,都比开着浏览器一点点刷要快。
4.2 测量值录入接口的校验与落库
录入接口最容易出问题的是校验缺失。前端传一个长度不对的数组进来,控制图就会被污染,所以校验必须在后端做死。
# routes/measure.py import json from datetime import datetime from flask import Blueprint, request, jsonify from models import db, Feature, Subgroup from spc_core import subgroup_stats measure_bp = Blueprint('measure', __name__) @measure_bp.route('/measurements', methods=['POST']) def add_measurement(): """录入一个子组:feature_code + 一组测量值 + 可选批次号""" data = request.get_json(force=True) feature = Feature.query.filter_by(code=data.get('feature_code')).first() if not feature: return jsonify({'code': 1, 'msg': '特性编码不存在'}), 404 values = data.get('values') if not isinstance(values, list) or len(values) < 2: return jsonify({'code': 1, 'msg': 'values 必须是长度不小于 2 的数组'}), 422 if len(values) != feature.subgroup_size: return jsonify({ 'code': 1, 'msg': f'子组容量应为 {feature.subgroup_size},收到 {len(values)}' }), 422 x_bar, rng = subgroup_stats(values) sg = Subgroup( feature_id=feature.id, batch_no=data.get('batch_no', ''), sample_time=parse_dt(data.get('sample_time')), values_json=json.dumps(values), mean=x_bar, rng=rng, ) db.session.add(sg) db.session.commit() return jsonify({'code': 0, 'msg': 'ok', 'x_bar': x_bar, 'r': rng})校验顺序很关键:先查特性存在性,再查数组类型和长度,最后才做统计计算。子组容量不匹配直接拒绝,这是最常见的接口误用场景——前端把采集批次长度从 5 改成 8,忘了同步特性配置,结果整张图标满报警。sample_time走独立解析函数,因为不同设备上报的格式不统一,有的带时区后缀,有的不带。
4.3 ECharts 渲染控制图:markLine 画控制限
图表渲染用 ECharts,原因是它对 markLine、dataZoom、tooltip 这些控制图必需元素支持最成熟。控制限三条线用markLine画,随接口返回的 limits 动态变化,不需要前端写死。
<!-- templates/chart.html --> <div id="xbar" style="width:100%;height:420px;"></div> <script src="{{ url_for('static', filename='js/echarts.min.js') }}"></script> <script> fetch('/spc/chart_data?feature_code=BZ-OD') .then(r => r.json()) .then(res => { const chart = echarts.init(document.getElementById('xbar')); chart.setOption({ title: { text: res.feature + ' Xbar 控制图' }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: res.points.map(p => p.sample_time) }, yAxis: { type: 'value', scale: true }, dataZoom: [{ type: 'inside' }, { type: 'slider' }], series: [{ name: 'Xbar', type: 'line', data: res.points.map(p => p.mean), markLine: { symbol: 'none', data: [ { yAxis: res.limits.x_ucl, name: 'UCL' }, { yAxis: res.limits.x_cl, name: 'CL' }, { yAxis: res.limits.x_lcl, name: 'LCL' } ], lineStyle: { type: 'dashed' } } }] }); }); </script>逻辑说明:yAxis不用写死 min/max,scale: true让坐标轴跟随数据范围,否则固定零点会把控制限压缩成一条直线。dataZoom加两个类型,滚轮缩放和底部滑块配合,几千个点也能流畅浏览。报警点可以用markPoint把支持规则命中的点标红,比只看表格直观得多。R 图画法完全一样,series 换成r字段,markLine 换成r_ucl和r_lcl。
4.4 点位膨胀后的查询与下采样
跑半年后单特性的子组数轻松上万,前端渲染和接口查询都会变慢。控制图的全部历史点其实没有必要一次画完,常见做法是近 200 个子组按原始点位返回,更早的历史按天聚合后只留均值趋势。
# routes/chart.py 内部 recent = (Subgroup.query .filter_by(feature_id=fid) .order_by(Subgroup.sample_time.desc()) .limit(200) .all()) older = (db.session.query( func.date(Subgroup.sample_time).label('d'), func.avg(Subgroup.mean).label('m'), func.avg(Subgroup.rng).label('r')) .filter(Subgroup.feature_id == fid) .filter(Subgroup.sample_time < recent[-1].sample_time) .group_by(func.date(Subgroup.sample_time)) .all())下采样的代价是边界处的判异规则会失真,因为聚合后的均值不再对应真实子组。处理办法是把“判异”职责从图表接口剥离出去,由定时任务在完整数据上计算并把报警写入RuleAlarm表,图表接口只负责展示已经判定过的结果。这样页面再快,也不会牺牲报警的准确性。
5. Gunicorn 部署与定时重算:SPC 软件上线的 3 个必调参数
5.1 Gunicorn 启动参数:worker 数和超时怎么定
flask run只配在开发环境用,它单进程处理请求,一个慢查询能把整个看板拖死。生产环境用 Gunicorn 当 Flask 服务器,启动命令和参数如下:
# 部署机上先确认 python3 版本和虚拟环境,再启动 gunicorn -w 4 -k sync -b 0.0.0.0:8080 --timeout 30 \ --access-logfile logs/access.log "app:create_app()"| 参数 | 推荐值 | 说明 |
|---|---|---|
| -w | 2 倍 CPU 核数 + 1 | 同步 worker 下一个进程同时处理一个请求 |
| -k | sync | 控制图计算是 CPU 密集任务,用协程 worker 反而引入调度开销 |
| --timeout | 30 | 超过 30 秒无响应的 worker 会被主进程杀掉 |
| -b | 0.0.0.0:8080 | 内网可访问,前面再挂 Nginx 做静态文件和 HTTPS 终结 |
timeout 要重点盯:控制限重算或全量判异如果放进请求里,超过 30 秒就会被杀。所以重活全部挪到定时任务,接口只读快照数据。Gunicorn 前面挂一层 Nginx,把 ECharts 的 js、css 交给 Nginx 直接返回,Flask 只出接口,静态文件不经过 Python 进程,这个分层能把页面响应压到百毫秒级。
5.2 控制限定时重算:APScheduler 与 cron 两种选法
控制限不是每次请求现算的,而是每天固定时间用前一天的全部子组重算,写入ControlLimitSnapshot表,当天接口只读快照。两种调度方式各有适用场景。
进程内的 APScheduler 适合“报警推送”这类随应用存活的任务:
# scheduler.py from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger def start_scheduler(app): sched = BackgroundScheduler() sched.add_job( func=lambda: recalc_limits(app), trigger=CronTrigger(hour=6, minute=0), id='daily_recalc', replace_existing=True, ) sched.start()但 Gunicorn 默认起 4 个 worker,每个 worker 都会执行start_scheduler,同一个定时任务跑 4 遍。解法是用环境变量只让一个进程启动,或者更干脆——把重算写成独立脚本,交给系统 cron 调度,应用内完全不碰调度器:
# crontab -e,每天 6 点半单独进程重算控制限 30 6 * * * cd /opt/spc && /opt/spc/venv/bin/python scripts/recalc_limits.py >> logs/recalc.log 2>&1独立脚本的好处是重算失败不影响 Web 进程,日志单独归档,排错路径清晰。SPC 这种每天一次的重算场景,用 cron 比在进程里塞调度器可靠得多。
5.3 排查三件事:迁移、时区、索引
上线后最常踩的三个坑,按出现频率排:
第一是表结构变更。db.create_all()只会建新表,不会改旧表,给spc_subgroup加列时必须上迁移工具:
flask db init flask db migrate -m "add spc_limit snapshot table" flask db upgrade第二是时间解析。datetime.fromisoformat在 Python 3.11 之前不接受带 Z 后缀的 ISO 时间串,很多设备上传的时间都带 Z。写个兼容函数:
def parse_dt(s): return datetime.fromisoformat(s.replace('Z', '+00:00'))第三是查询计划。子组表动辄几十万行,控制图接口按feature_id + sample_time过滤,必须命中复合索引。models.py里的__table_args__定义的ix_feature_time就是这个用途。加上复合索引和控制限快照之后,图表接口的响应时间不再随历史点数线性变慢,查询计划能稳定走索引覆盖,大促期间哪怕同时开几十个看板也不会把 MySQL 打满。
本文还有配套的精品资源,点击获取