# -*- coding: utf-8 -*- """压测(M7-09):门禁并发 / 上传并发 / 读接口 / 导出大任务 ## 为什么值得真跑一次,而不是"看代码觉得没问题" 这个项目的三个并发热点,每一个都有"看代码看不出来"的地方: | 场景 | 直觉 | 实际可能卡在 | |---|---|---| | 门禁核验(60 人同时扫码) | "就是查一下名册" | 每个请求都要 **AES 解密全表证件号** 逐个比对(密文没法走索引) | | 上传(30 人同时传) | "就是写个文件" | 图片压缩是 **CPU 密集**,而 Tomcat 线程数有限 | | 读接口(列表/统计) | "加个索引就行" | 连接池只有 20 条,长事务(导出打包)会把池子占死 | | 导出大任务 | "异步的,不影响" | 导出**同步**做前置检查 + 打包,而它在一个大事务里 | 所以这份脚本不做"泛泛压一下",而是**针对这四个场景各设计一组**, 并把 p50/p95/p99 与错误码分布打出来 —— **错误码分布比成功率更有信息量** (40071 是限流、50000 是内部错、40301 是权限,三种要修的地方完全不同)。 ## 用法 ```bash python scripts/load-test.py # 全套(约 2~3 分钟) python scripts/load-test.py --only gate # 只跑门禁 python scripts/load-test.py --conc 40 # 改并发 ``` ⚠️ 它**会往库里写东西**(名册、填报单、材料),结尾会清理; 中途 Ctrl-C 会留下残留 —— 用 `--clean` 单独清理一次即可。 """ import argparse import base64 import concurrent.futures as cf import importlib.util import json import os import statistics import subprocess import sys import time import urllib.error import urllib.request import uuid sys.stdout.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # 默认打 dev 后端;`--base` 可改(例如为了量上传而在 18080 另起一个 # `--fill.storage.provider=local` 的实例,避免把测试文件灌进真实对象存储) BASE = "http://localhost:8080" # 演示班主任(scripts/dev-seed-audit.py 摆出来的)—— 一期初审恒为开启, # 所以"审批闭环"这类场景必须有它 TEACHER_USER = "demot1" TEACHER_PWD = "Demo1234" SUFFIX = uuid.uuid4().hex[:6] TMP_ROSTER = os.path.join(ROOT, "scripts", f"_lt-roster-{SUFFIX}.xlsx") TMP_IMG = os.path.join(ROOT, "scripts", f"_lt-img-{SUFFIX}.jpg") # 压测规模(可按机器调整;--rows 会覆盖 GATE_USERS) GATE_USERS = 60 # 名册人数(同时也是门禁的并发用户数) READ_CONC = 20 # 读接口并发 READ_ROUNDS = 8 # 每个并发线程的轮数 UPLOAD_USERS = 20 # 上传:20 个学生同时传 UPLOAD_PER_USER = 2 # 每人传几张 TASK_IDS = [] ROSTER_IDS = [] UIDS = [] def _load(name, fn): spec = importlib.util.spec_from_file_location(name, os.path.join(ROOT, "scripts", fn)) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod R = _load("e2e_m2_roster", "e2e-m2-roster.py") def sql(s): return R.sql(s)[1] def one(s): v = sql(s) return v.split("\n")[0].strip() if v else "" def call(method, path, token=None, body=None, timeout=60): return R.call(method, BASE + path, token, body, timeout=timeout) # ══════════════ 计时与统计 ══════════════ class Stat: """一个场景的耗时与错误码统计。""" def __init__(self, name): self.name = name self.lat = [] self.codes = {} self.fail_detail = [] self.t0 = time.perf_counter() self.t1 = None def done(self): """场景结束(算吞吐要墙钟时间,不是"平均耗时的倒数")。""" self.t1 = time.perf_counter() def add(self, ms, code): self.lat.append(ms) self.codes[code] = self.codes.get(code, 0) + 1 def report(self): if not self.lat: print(" %-22s 没有样本" % self.name) return s = sorted(self.lat) n = len(s) ok = self.codes.get(0, 0) # ⚠️ 吞吐 = 请求数 / **墙钟耗时**。 # 原来的式子化简后是 1/平均耗时 —— 那不是吞吐, # 而"一行看起来很像结论的数字"最容易被抄进汇报里 elapsed = (self.t1 or time.perf_counter()) - self.t0 print(" %-24s 请求 %-5d 成功 %-5d 失败 %-4d " "p50 %6.1fms p95 %7.1fms p99 %7.1fms max %7.1fms 吞吐 %5.0f/s" % (self.name, n, ok, n - ok, s[int(n * 0.50)], s[min(n - 1, int(n * 0.95))], s[min(n - 1, int(n * 0.99))], s[-1], n / max(0.001, elapsed))) # ⚠️ 错误码分布比成功率更有信息量:40071 限流 / 50000 内部错 / 40301 权限 bad = {k: v for k, v in self.codes.items() if k != 0} if bad: print(" 失败分布:%s" % json.dumps(bad, ensure_ascii=False)) if self.fail_detail: print(" 首个失败样本:%s" % self.fail_detail[0][:200]) def timed(fn): t0 = time.perf_counter() try: code, body = fn() except Exception as e: # noqa: BLE001 return (time.perf_counter() - t0) * 1000, -1, str(e) return (time.perf_counter() - t0) * 1000, code, body # ══════════════ 夹具 ══════════════ def uid_of(token): """ 从自己的 token 里取 uid(`sub`)。 ⚠️ **不要按 username 查库**:库里同一个 username 在每个租户都有一行 (`admin` 有 50 多个,`select id from sys_user where username='admin'` 会随机命中 别人家的账号)。我第一版就是这么写的,于是把"初审人"设成了一个跨租户的用户 id。 """ p = token.split(".")[1] p += "=" * (-len(p) % 4) return int(json.loads(base64.urlsafe_b64decode(p))["sub"]) def call_ok(method, path, token, body=None, what="", timeout=60): """打一个接口并**断言业务码为 0**(不查返回码的调用是 F-96 里最贵的那一个)。""" c, b = call(method, path, token, body, timeout=timeout) if not (isinstance(b, dict) and b.get("code") == 0): raise SystemExit("夹具失败:%s → %s" % (what or path, json.dumps(b, ensure_ascii=False)[:200])) return b.get("data") def build_fixture(admin, teacher): """造一个任务 + N 个名册人(每人一个已知证件号)+ 对应 uid 的 MP token。""" _, orgs = call("GET", "/api/v1/orgs/tree", admin) classes = [] def walk(ns): for n in ns or []: if n.get("orgType") == 3: classes.append(n) walk(n.get("children")) walk(orgs.get("data") or []) if not classes: raise SystemExit("没有班级节点") org = classes[0] _, b = call("POST", "/api/v1/tasks", admin, { "taskName": f"压测任务 {SUFFIX}", "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2099-12-31", # ⚠️ **只走终审**(needClassAudit=false)是刻意的:初审的判据是"这份单在你负责的 # 班级里"(AuditService.auditDenyReason),而夹具的名册导进的是组织树里第一个班, # admin 并不是那个班的班主任 —— 于是逐份通过会整齐地报 40301「不在你负责的班级里」。 # 要按真实两段审核走,就得再造一个"带班级范围的班主任"账号; # 而导出这条路与"谁审的"无关,所以这里选便宜且不引入额外夹具的那条。 "auditConfig": {"needClassAudit": False, "needAdminAudit": True}, }) tid = b["data"]["id"] TASK_IDS.append(tid) call("POST", f"/api/v1/tasks/{tid}/materials", admin, { "materialCode": "IDCARD", "materialName": "身份证", "required": True, "fileTypes": ["jpg", "jpeg", "png"], "maxSizeKb": 4096, "maxCount": 3, "minSizeKb": 1, "minLongSide": 100, }) # ⚠️ **必须指定 headUserId**:初审的判据是"这个班归不归我审" # (`AuditScopeMapper.countMyOrg` 三级回退:head_user_id → deputy → 班级负责人), # 而夹具的名册导进的是组织树里第一个班 —— 不指定的话,逐份通过会整齐地报 # 40301「这份填报单不在你负责的班级里」。 # 也**不能**靠"关掉初审"绕过:一期 `withAuditDefaults` 把 needClassAudit 强制为 true # (契约 §3.2「社会收集恒为 false,一期不启用社会收集」),传 false 会被扳回来。 # 指定的人还得**真有班主任角色**,否则这次 PUT 会被服务端拒(我第一版就是被它拒的, # 而当时没查返回码,表现成很远的地方"任务已截止"—— 40008)。 call_ok("PUT", f"/api/v1/tasks/{tid}/orgs", admin, {"orgs": [{"orgId": org["id"], "expectedCount": GATE_USERS, "headUserId": uid_of(teacher)}]}, "设置覆盖班级与初审人") # N 个证件号(前 17 位递增,校验位由脚本补) rows = [] for i in range(GATE_USERS): # ⚠️ 身份证号是**定长**的:6 位地区 + 8 位出生日期 + 3 位顺序 + 1 位校验。 # 所以任何一段"从序号推出来的值"都必须**取模保证位数** —— # 我原来写 `"%03d" % (100 + i)`,i >= 900 时变成 4 位, # 号码成了 19 位,2000 人的夹具只导进去 900 个。 # # 这里把序号拆成"地区码分块 + 3 位顺序号":每 1000 人换一个地区码。 # 出生日期固定用 20080101(必须是**真实存在的日期** —— 我曾经拼出 # `20081000`,被导入校验以"出生日期要真实存在"拒掉,而那条校验抓的 # 不是格式,是"这个号码不可能属于一个人")。 area = "%06d" % (361029 + i // 1000) seq = "%03d" % (i % 1000) idc = R.with_checksum(area + "20080101" + seq) rows.append((f"压测{i:04d}", idc)) t_excel = time.perf_counter() build_roster(rows) excel_s = time.perf_counter() - t_excel t_imp = time.perf_counter() body = R.upload_roster(BASE, admin, tid, TMP_ROSTER, extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]}) imp_s = time.perf_counter() - t_imp # ⚠️ 两段分开报:**造 Excel 是压测脚本自己的成本,不是系统的成本** —— # 混在一起会把 openpyxl 的耗时算到"名册导入"头上(度量工具的偏差伪装成产品缺陷) print(" 夹具计时:造 Excel %.1fs 导入接口 %.1fs(%d 行)" % (excel_s, imp_s, len(rows))) # ⚠️ **判据是 data.created,不是 code**:导入了 0 行也是 code=0 # (体检报告在 data 里)。判错的话压测会"没有样本", # 而那个现象看起来像压测脚本坏了 —— 与 F-84 是同一个坑 rep = body.get("data") or {} if body.get("code") != 0 or rep.get("created") != GATE_USERS: raise SystemExit("夹具名册导入没达标:code=%s created=%s errors=%s" % (body.get("code"), rep.get("created"), json.dumps(rep.get("errors"), ensure_ascii=False)[:300])) ids = sql(f"select id, id_card_no from collect_roster where task_id={tid} " f"and deleted=0 order by id;") # ⚠️ 证件号在库里是密文,压测要的是**明文**(门禁接口收明文),所以直接用生成的 roster_rows = [] for idx, line in enumerate([x for x in ids.split("\n") if x.strip()]): rid = int(line.split("|")[0]) ROSTER_IDS.append(rid) roster_rows.append({"rosterId": rid, "name": rows[idx][0], "idCardNo": rows[idx][1]}) if len(roster_rows) != GATE_USERS: raise SystemExit("夹具名册行数不对:期望 %d,实际 %d" % (GATE_USERS, len(roster_rows))) # ⚠️ 发布**必须查返回码**:它失败时任务还是草稿,而后续所有学生接口都报 # `40008 任务已截止、已归档或已暂停` —— 排查会一路查错方向(我踩过) call_ok("POST", f"/api/v1/tasks/{tid}/publish", admin, {"force": True}, "发布任务") cfg = R.read_dev_config() tokens = [] for i, r in enumerate(roster_rows): uid = 998000 + i UIDS.append(uid) tokens.append({"uid": uid, "token": R.mint_mp_token( cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4), "name": r["name"], "idCardNo": r["idCardNo"], "rosterId": r["rosterId"]}) return tid, tokens def build_roster(rows): from openpyxl import Workbook wb = Workbook() ws = wb.active ws.title = "压测班" header = ["序号", "姓名", "身份证号", "学历", "联系电话", "工作单位", "报考职业", "报考工种", "报考等级", "是否补考", "补考理论", "补考实操", "补考综合", "是否在校生", "在读学校", "所学专业", "入学时间"] for c, t in enumerate(header, start=1): ws.cell(row=1, column=c, value=t) for i, (name, idc) in enumerate(rows): vals = [i + 1, name, idc, "中专", "138001399%02d" % (i % 100), "无", "车工", "数控车工", "四级", "否", "", "", "", "是", "示范职业学校", "数控加工", "2024-09-01"] for c, v in enumerate(vals, start=1): if v != "": ws.cell(row=i + 2, column=c, value=v) wb.save(TMP_ROSTER) def make_img(size_kb=400, px=(2000, 1500)): """ 造一张**真实**的 JPEG(Pillow 噪声图)。 ⚠️ 原来是"JPEG 头 + 随机字节"的假图。它能过扩展名与大小校验, 但**服务端做的根本不是同一件事**:真图要解码 → 按 EXIF 转正 → 重编码, 假图在解码那一步就失败了(然后走的是"压缩失败"的分支)。 用压测去回答"压缩是 CPU 密集",却喂它一个不需要压缩的文件 —— 量到的不是压缩成本,而是错误处理的成本。**这类"夹具不对"的红/绿都不算数。** 噪声图是有意选的:纯色图会被压到极小,量不出 CPU 成本 (与 e2e-m9-thumb.py 里那条"纯色会被压得很小、失真"同一个理由)。 """ try: from PIL import Image import random random.seed(7) img = Image.new("RGB", px) img.putdata([(random.randrange(256), random.randrange(256), random.randrange(256)) for _ in range(px[0] * px[1])]) img.save(TMP_IMG, "JPEG", quality=92) return os.path.getsize(TMP_IMG) except ImportError: import random random.seed(7) with open(TMP_IMG, "wb") as f: f.write(b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00") f.write(bytes(random.getrandbits(8) for _ in range(size_kb * 1024))) f.write(b"\xff\xd9") return os.path.getsize(TMP_IMG) def reset_gate(): """清门禁限流(每用户 10 次/小时)。压测每个用户只核验 1 次,理论上不触发; 但重复跑会累积,所以开跑前清一次。""" for uid in UIDS: os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:mp:gate:user:%d > NUL 2>&1' % uid) # ══════════════ 场景 ══════════════ def scenario_gate(tid, tokens, conc): """门禁核验:N 个学生同时扫码。""" st = Stat("门禁核验") reset_gate() def one_req(t): def go(): c, b = call("POST", "/api/v1/mp/roster/match", t["token"], {"taskId": tid, "realName": t["name"], "idCardNo": t["idCardNo"]}, timeout=60) return c if c == 200 else c, b ms, code, body = timed(go) # 业务码在 body 里(HTTP 200 + code != 0),要分开统计 biz = body.get("code") if isinstance(body, dict) else None return ms, int(biz if biz is not None else code), body with cf.ThreadPoolExecutor(max_workers=conc) as ex: for ms, code, body in ex.map(one_req, tokens): st.add(ms, code) if code != 0 and len(st.fail_detail) < 3: st.fail_detail.append(json.dumps(body, ensure_ascii=False)) st.done() st.report() return st def scenario_read(tid, admin, conc, rounds): """ 读接口:任务列表 / 名册列表 / 任务统计 / 待办。 ⚠️ **按端点分开统计**,不要只报一个"混合 p95": 压测的目的就是回答"该优化哪个",而混在一起的数字回答不了 —— 四个接口里可能有一个占掉 90% 的耗时,而混合 p95 会把它的形状抹平。 """ paths = [ ("GET", f"/api/v1/tasks?pageNum=1&pageSize=20", "任务列表"), ("GET", f"/api/v1/tasks/{tid}/roster?pageNum=1&pageSize=50", "名册列表(50/页)"), ("GET", f"/api/v1/tasks/{tid}/roster?pageNum=1&pageSize=200", "名册列表(200/页)"), ("GET", f"/api/v1/tasks/{tid}/statistics", "任务统计"), ("GET", f"/api/v1/audit/todos?pageNum=1&pageSize=20", "审核待办"), ] stats = {label: Stat(label) for _m, _p, label in paths} def worker(_): out = [] for _r in range(rounds): for m, p, label in paths: def go(): c, b = call(m, p, admin) return c, b ms, code, body = timed(go) biz = body.get("code") if isinstance(body, dict) else None out.append((label, ms, int(biz if biz is not None else code), body)) return out with cf.ThreadPoolExecutor(max_workers=conc) as ex: for results in ex.map(worker, range(conc)): for label, ms, code, body in results: st = stats[label] st.add(ms, code) if code != 0 and len(st.fail_detail) < 3: st.fail_detail.append(json.dumps(body, ensure_ascii=False)) for st in stats.values(): st.done() for label in [x[2] for x in paths]: stats[label].report() return stats def scenario_upload(tid, tokens, conc, px=(2000, 1500)): """上传:并发上传图片(压缩是 CPU 密集)。""" st = Stat("材料上传") size = make_img(px=px) print(" 夹具:真实 JPEG %d×%d / %.0f KB" % (px[0], px[1], size / 1024.0)) # ⚠️ 前置条件:**必须先把人绑到名册项上**,否则 `POST /mp/submissions` 会被拒 # ("没有草稿"),而 40 次上传会整齐地失败 —— 那不叫"上传慢",叫"夹具没成立"。 # 以前单独跑 `--only upload` 就是这个结果:全量压测里它跟在门禁后面, # "恰好"能跑,于是这个前置依赖一直没被发现(F-96 的同一类:没断言前置就往下走)。 # 绑定逻辑与导出大包共用 `bind_users` —— 写两份必然分叉,而分叉的表现就是 # "另一个场景的前置悄悄失效"。 bound = bind_users(tid, tokens, conc) print(" 前置:门禁绑定 %d/%d 人(上传必须先有绑定)" % (bound, len(tokens))) if bound < conc: raise SystemExit("上传压测的前置条件不成立:只有 %d 人绑定成功,需要 %d 人" % (bound, conc)) sub_ids = {} for t in tokens[:conc + 1]: # 多要一份:串行基线占用第 0 个 c, b = call("POST", "/api/v1/mp/submissions", t["token"], {"taskId": tid}) if c == 200 and isinstance(b, dict) and b.get("code") == 0: sub_ids[t["uid"]] = b["data"]["id"] else: print(" ⚠️ 建草稿失败:%s" % json.dumps(b, ensure_ascii=False)[:160]) # ⚠️ 同样要断言:建不出草稿时"40 个请求全失败"看起来像上传功能坏了 if len(sub_ids) < conc + 1: raise SystemExit("上传压测的前置条件不成立:只建出 %d 份草稿,需要 %d 份" % (len(sub_ids), conc + 1)) # ── 串行基线:单请求到底多贵 ── # ⚠️ 必须有这一段。并发下的 p50/p95 里混着排队,**不能反推单请求成本**(F-94); # 而"值不值得优化"只能由"单请求成本 vs 排队占比"回答。 warm = tokens[0] w_ms, w_http, w_body = timed(lambda: upload_file( "/api/v1/mp/files/upload", warm["token"], sub_ids[warm["uid"]], TMP_IMG)) # ⚠️ 判据是 body 里的业务码,**不是** upload_file 返回的第一个值 —— # 那是 HTTP 状态码(200),拿它当"成功"判据会把成功报成失败(我自己刚踩过) if isinstance(w_body, dict) and w_body.get("code") == 0: # 尺寸以库里那行为准:接口回的是原始大小,规范件大小只有落库才知道 sizes = one("select raw_size || '/' || coalesce(final_size, 0) from collect_file " "where submission_id=%d order by id desc limit 1;" % sub_ids[warm["uid"]]) try: raw_kb, fin_kb = [int(x) / 1024.0 for x in sizes.split("/")] ratio = (fin_kb / raw_kb * 100.0) if raw_kb else 0.0 print(" 串行基线:单请求 %.0fms 压缩 %.0fKB → %.0fKB(%.0f%%,服务端真的在压)" % (w_ms, raw_kb, fin_kb, ratio)) except Exception: # noqa: BLE001 print(" 串行基线:单请求 %.0fms(尺寸没取到:%s)" % (w_ms, sizes)) else: print(" 串行基线失败:HTTP %s %s" % (w_http, json.dumps(w_body, ensure_ascii=False)[:160])) def one_req(t): sid = sub_ids.get(t["uid"]) if sid is None: return 0.0, -2, {"msg": "没有草稿"} def go(): return upload_file("/api/v1/mp/files/upload", t["token"], sid, TMP_IMG) ms, code, body = timed(go) biz = body.get("code") if isinstance(body, dict) else None return ms, int(biz if biz is not None else code), body jobs = [] # 并发用 tokens[1:conc+1]:第 0 个已经被串行基线用掉了,混进来会让"并发里的某个用户 # 比别人多传一次",而样本数对不上时最难判断的是"到底谁慢" for t in tokens[1:conc + 1]: for _ in range(UPLOAD_PER_USER): jobs.append(t) with cf.ThreadPoolExecutor(max_workers=conc) as ex: for ms, code, body in ex.map(one_req, jobs): st.add(ms, code) if code != 0 and len(st.fail_detail) < 3: st.fail_detail.append(json.dumps(body, ensure_ascii=False)) st.done() st.report() return st def upload_file(path, token, submission_id, file_path, timeout=180): boundary = "----lt" + uuid.uuid4().hex[:12] name = os.path.basename(file_path) with open(file_path, "rb") as f: content = f.read() def part(field, value): return (f"--{boundary}\r\nContent-Disposition: form-data; name=\"{field}\"\r\n\r\n" f"{value}\r\n").encode("utf-8") body = b"" body += part("submissionId", submission_id) body += part("materialCode", "IDCARD") body += (f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; " f"filename=\"{name}\"\r\nContent-Type: image/jpeg\r\n\r\n").encode("utf-8") body += content + b"\r\n" + f"--{boundary}--\r\n".encode("utf-8") req = urllib.request.Request(BASE + path, data=body, method="POST", headers={ "Content-Type": f"multipart/form-data; boundary={boundary}", "Authorization": f"Bearer {token}", }) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: return e.code, {"msg": e.read().decode("utf-8", "replace")[:200]} except Exception as e: # noqa: BLE001 return -1, {"msg": str(e)} # ══════════════ 长时间稳定性 ══════════════ def base_port(): from urllib.parse import urlparse return urlparse(BASE).port or 80 def app_pid(port): """监听该端口的进程 pid(没有就返回 None,不猜)。""" try: r = subprocess.run(["powershell", "-NoProfile", "-Command", "(Get-NetTCPConnection -State Listen -LocalPort %d " "-ErrorAction SilentlyContinue).OwningProcess" % port], capture_output=True, text=True, timeout=30) for tok in (r.stdout or "").split(): if tok.strip().isdigit(): return int(tok.strip()) except Exception: # noqa: BLE001 pass return None def rss_mb(pid): """ 进程的物理内存(MB)。 ⚠️ 从**系统**取,不从应用的指标取:应用自己报的堆用量看不到堆外内存、 元空间与线程栈 —— 而"内存慢慢涨"恰恰经常长在这些地方。 """ if not pid: return None try: r = subprocess.run(["powershell", "-NoProfile", "-Command", "(Get-Process -Id %d -ErrorAction SilentlyContinue).WorkingSet64" % pid], capture_output=True, text=True, timeout=30) v = (r.stdout or "").strip() return int(v) / 1048576.0 if v.isdigit() else None except Exception: # noqa: BLE001 return None def db_conns(): """PG 上到 filldb 的连接数 —— Hikari 连接泄漏会直接体现为它单调上升。""" v = one("select count(*) from pg_stat_activity where datname='filldb';") return int(v) if v.isdigit() else None def scenario_soak(tid, admin, minutes, rps): """ 长时间稳定性:低强度持续打,每分钟采样一行。 ⚠️ 要回答的是"**有没有单调上升**",所以判据是**趋势**,不是"跑完没崩"。 泄漏的特征是"慢慢地越来越高",一次性压测永远测不出来 —— 所以这里每分钟记 RSS 与 PG 连接数,最后比较**前 1/3 与后 1/3** 的均值。 延迟趋势一起看:泄漏的另一个表现是"越跑越慢"。 """ paths = [ ("GET", "/api/v1/tasks?pageNum=1&pageSize=20", "任务列表"), ("GET", f"/api/v1/tasks/{tid}/roster?pageNum=1&pageSize=50", "名册列表"), ("GET", f"/api/v1/tasks/{tid}/statistics", "任务统计"), ("GET", "/api/v1/audit/todos?pageNum=1&pageSize=20", "审核待办"), ] pid = app_pid(base_port()) print(" 目标后端 pid=%s(RSS 从系统取,PG 连接数从 pg_stat_activity 取)" % pid) print(" 强度 %.1f 请求/秒 时长 %d 分钟,每分钟一行样本" % (rps, minutes)) rows = [] n = 0 t_end = time.time() + minutes * 60 while time.time() < t_end: lat, fail = [], 0 for _ in range(max(1, int(rps * 60))): m, p, _label = paths[n % len(paths)] n += 1 ms, code, body = timed(lambda: call(m, p, admin)) biz = int((body or {}).get("code", code)) if isinstance(body, dict) else code lat.append(ms) if biz != 0: fail += 1 # 均匀铺开:不要"一分钟里的前 10 秒打完全部请求",那测的是尖峰不是持续压力 time.sleep(max(0.0, 1.0 / rps)) s = sorted(lat) row = {"min": len(rows) + 1, "n": len(lat), "fail": fail, "p50": s[len(s) // 2], "p95": s[min(len(s) - 1, int(len(s) * 0.95))], "rss": rss_mb(pid), "conns": db_conns()} rows.append(row) print(" %3d 分 请求 %-4d 失败 %-3d p50 %6.1fms p95 %7.1fms RSS %7s MB PG 连接 %4s" % (row["min"], row["n"], row["fail"], row["p50"], row["p95"], ("%.0f" % row["rss"]) if row["rss"] else "—", row["conns"] if row["conns"] is not None else "—")) def avg(rs, key): vals = [r[key] for r in rs if r.get(key) is not None] return sum(vals) / len(vals) if vals else None third = max(1, len(rows) // 3) head, tail = rows[:third], rows[-third:] print("\n 样本 %d 分钟 请求 %d 失败 %d(%s)" % (len(rows), sum(r["n"] for r in rows), sum(r["fail"] for r in rows), "强度 %.1f/s、规模 %d 行" % (rps, GATE_USERS))) for key, unit, label, tol in (("rss", "MB", "RSS", 10.0), ("conns", "个", "PG 连接数", 20.0)): a, b = avg(head, key), avg(tail, key) if a is None or b is None: print(" %-10s 样本不足,没测出来" % label) continue pct = ((b - a) / a * 100.0) if a else 0.0 verdict = "平稳" if abs(pct) < tol else ("**上升** ⚠️ 疑似泄漏" if b > a else "下降") print(" %-10s 前 1/3 均值 %8.1f %s → 后 1/3 均值 %8.1f %s(%+.1f%%) %s" % (label, a, unit, b, unit, pct, verdict)) a95, b95 = avg(head, "p95"), avg(tail, "p95") if a95 and b95: print(" %-10s 前 1/3 均值 %8.1f ms → 后 1/3 均值 %8.1f ms(%+.1f%%)" % ("延迟 p95", a95, b95, (b95 - a95) / a95 * 100.0)) return rows def bind_users(tid, tokens, workers=10): """ 把 N 个人绑到名册项上(上传 / 提交 / 审核的**共同前置**)。 ⚠️ 抽出来是有原因的:只在 `scenario_upload` 里写一份的话,新场景(导出大包) 就会忘 —— 而忘了的表现是"请求整齐地全失败",看起来像功能坏了(F-96 的同一类)。 """ ok = 0 fail_sample = None def one(t): nonlocal fail_sample c, b = call("POST", "/api/v1/mp/roster/match", t["token"], {"taskId": tid, "realName": t["name"], "idCardNo": t["idCardNo"]}) code = b.get("code") if isinstance(b, dict) else -1 if code not in (0, 40072) and fail_sample is None: # 0 绑定成功 / 40072 已绑定 fail_sample = json.dumps(b, ensure_ascii=False)[:160] return code with cf.ThreadPoolExecutor(max_workers=workers) as ex: for code in ex.map(one, tokens): if code in (0, 40072): ok += 1 if fail_sample is not None: print(" ⚠️ 绑定失败样本:%s" % fail_sample) return ok def scenario_export_big(tid, tokens, admin, teacher, files_per_user=1, split_mb=0, conc=10): """ 导出**大包**:造 N 份已通过的填报单,量打包墙钟、体积、分卷与产物的完整性。 这是"交付物本身"的那条链,也是唯一一条**没法用接口断言替代**的验证: 打包耗时、产物体积、分卷是否每卷都能单独解开,只有真跑一遍才知道。 """ import zipfile n = len(tokens) print(" 目标:%d 人 × %d 个材料%s" % (n, files_per_user, ",分卷 %dMB" % split_mb if split_mb else ",不分卷")) bound = bind_users(tid, tokens, conc) print(" ① 门禁绑定 %d/%d 人" % (bound, n)) if bound < n: print(" ⚠️ 有 %d 人没绑上 —— 后面的上传会失败,先看失败码再判断是不是导出的问题" % (n - bound)) sid = {} draft_fail = None def draft(t): nonlocal draft_fail c, b = call("POST", "/api/v1/mp/submissions", t["token"], {"taskId": tid}) if not (isinstance(b, dict) and b.get("code") == 0): # ⚠️ 失败**必须打出来**:上一次它只有一句"草稿 0/150", # 排查时完全看不出是绑定没生效、还是任务状态不对(F-96 的同一类) if draft_fail is None: draft_fail = json.dumps(b, ensure_ascii=False)[:200] return t["uid"], None return t["uid"], (b.get("data") or {}).get("id") with cf.ThreadPoolExecutor(max_workers=conc) as ex: for uid, s in ex.map(draft, tokens): if s: sid[uid] = s print(" ② 草稿 %d/%d 份%s" % (len(sid), n, "" if draft_fail is None else " 首个失败:%s" % draft_fail)) if len(sid) < n: raise SystemExit("草稿没建齐(%d/%d),导出大包的夹具不成立" % (len(sid), n)) img_kb = make_img() / 1024.0 jobs = [(t, k) for t in tokens if t["uid"] in sid for k in range(files_per_user)] def up(job): t, _k = job _c, b = upload_file("/api/v1/mp/files/upload", t["token"], sid[t["uid"]], TMP_IMG) return (b or {}).get("code") if isinstance(b, dict) else -1 t_up = time.perf_counter() up_fail = 0 with cf.ThreadPoolExecutor(max_workers=conc) as ex: for code in ex.map(up, jobs): if code != 0: up_fail += 1 print(" ③ 上传 %d 张(失败 %d;%.0fKB/张;%.1fs)" % (len(jobs), up_fail, img_kb, time.perf_counter() - t_up)) if up_fail: raise SystemExit("上传有 %d 张失败,夹具不成立(先看失败码,别当成导出的问题)" % up_fail) def submit(t): c, b = call("POST", "/api/v1/mp/submissions/%d/submit" % sid[t["uid"]], t["token"], {"confirmWarnings": True}) # ⚠️ 判据是 data.submitted(F-84):HTTP 200 + code 0 不等于"提交成功" return ((b or {}).get("data") or {}).get("submitted") if isinstance(b, dict) else False ok_sub = 0 with cf.ThreadPoolExecutor(max_workers=conc) as ex: for v in ex.map(submit, [t for t in tokens if t["uid"] in sid]): if v is True: ok_sub += 1 print(" ④ 提交 %d/%d 份(判据 data.submitted)" % (ok_sub, len(sid))) if ok_sub != len(sid): raise SystemExit("有 %d 份没提交成功,夹具不成立" % (len(sid) - ok_sub)) ids = [sid[t["uid"]] for t in tokens if t["uid"] in sid] def approve(sid_, token, body_, label): c, b = call("POST", "/api/v1/audit/submissions/%d/approve" % sid_, token, body_, timeout=120) return (b or {}).get("code") if isinstance(b, dict) else -1, b # ⚠️ 一期**恒有两段审核**(needClassAudit 被强制 true,见 build_fixture 的说明), # 所以"逐份通过"要走两次:班主任初审 → 管理员终审。 # 只做一段的表现是全部停在 40301「不在你负责的班级里」或"没有待你处理的审核动作"。 for stage, (tok, body_, label) in enumerate(( (teacher, {"actionSource": "BUTTON", "remark": "压测初审"}, "初审"), (admin, {"actionSource": "BUTTON", "finalOpinion": "压测终审"}, "终审"))): ok_n, first_fail = 0, None with cf.ThreadPoolExecutor(max_workers=conc) as ex: for code, b in ex.map(lambda s: approve(s, tok, body_, label), ids): if code == 0: ok_n += 1 elif first_fail is None: first_fail = b print(" ⑤-%d %s通过:%d/%d 份%s" % (stage + 1, label, ok_n, len(ids), "" if first_fail is None else " 首个失败:%s" % json.dumps(first_fail, ensure_ascii=False)[:160])) if first_fail is not None: raise SystemExit("%s这一关没过,夹具不成立" % label) _c, b = call("GET", "/api/v1/exports/precheck?taskId=%d&onlyApproved=true" % tid, admin) pre = (b or {}).get("data") or {} print(" ⑥ 前置检查:canExport=%s 候选 %s 人 / %s 个文件 / 原件 %.1f MB" % (pre.get("canExport"), pre.get("candidateCount"), pre.get("fileCount"), (pre.get("rawSizeBytes") or 0) / 1048576.0)) if pre.get("canExport") is not True: raise SystemExit("前置检查不通过,大包测不了:%s" % json.dumps(pre.get("blocking"), ensure_ascii=False)[:200]) body = {"taskId": tid, "exportName": "压测大包 %s" % SUFFIX, "scope": {"onlyApproved": True}} if split_mb: body["splitSizeMb"] = split_mb t0 = time.perf_counter() _c, b = call("POST", "/api/v1/exports", admin, body, timeout=600) eid = ((b or {}).get("data") or {}).get("exportId") if (b or {}).get("code") != 0 or not eid: raise SystemExit("创建导出失败:%s" % json.dumps(b, ensure_ascii=False)[:200]) print(" ⑦ 导出 id=%s,创建耗时 %.1fs → 轮询到完成(打包是异步的,量墙钟)" % (eid, time.perf_counter() - t0)) detail, seen = None, None while time.perf_counter() - t0 < 1800: _c, b = call("GET", "/api/v1/exports/%d" % eid, admin) d = (b or {}).get("data") or {} if d.get("status") in (2, 3, 4): detail = d break if d.get("progress") != seen: seen = d.get("progress") print(" …%s%% %s" % (seen, d.get("stepText") or "")) time.sleep(2) wall = time.perf_counter() - t0 if not detail or detail.get("status") != 2: print(" ✗ 导出没跑完:status=%s errorMsg=%s" % (detail and detail.get("status"), detail and detail.get("errorMsg"))) return None s = detail.get("stats") or {} print(" ⑧ **打包墙钟 %.1fs** 文件 %s 个 原件 %.1f MB → 产物 %.1f MB(%s)" % (wall, s.get("fileCount"), s.get("rawSizeMb") or 0, s.get("finalSizeMb") or 0, s.get("compressRatio"))) # ⚠️ 列名以 DDL 为准(V11):`part_no` / `file_name` / `file_size` / `file_path` —— # 按直觉写成 part/name/size/path 会直接 SQL 报错(我第一版就是这么写的) vols = sql("select part_no, file_name, coalesce(file_size,0), file_path " "from collect_export_file where export_id=%d and deleted=0 order by part_no;" % eid) rows = [x for x in vols.split("\n") if x.strip()] print(" ⑨ 分卷 %d 个:" % len(rows)) for line in rows: part, name, size, path = (line.split("|") + ["", "", "", ""])[:4] print(" %s 卷 %s %.1f MB" % (name, part, int(size or 0) / 1048576.0)) # ⚠️ 分卷的承诺是"**每一卷都是完整可解的 ZIP**"(不是 .z01 那种必须凑齐才能解)。 # 这条只能真去解一遍 —— 断言"文件存在"是断言不到它的。 base = os.path.join(ROOT, "backend", "data", "uploads") for line in rows: part, name, size, path = (line.split("|") + ["", "", "", ""])[:4] full = path if os.path.isabs(path) else os.path.join(base, path) if not os.path.exists(full): # 存储是七牛时产物也可能不在本地(D5 说导出包落本地盘,所以这条算异常) print(" ? %s 不在本地磁盘(路径 %s)" % (name, full)) continue try: with zipfile.ZipFile(full) as z: names = z.namelist() bad = z.testzip() print(" ✓ %s 可解:%d 个条目,首条 %s%s" % (name, len(names), names[0] if names else "(空)", "" if bad is None else " ✗ CRC 错误:%s" % bad)) except Exception as e: # noqa: BLE001 print(" ✗ %s **解不开**:%s" % (name, e)) return detail def scenario_export(tid, admin): """导出:前置检查(只读)+ 创建(同步打包)。""" st_pre = Stat("导出前置检查") ms, code, body = timed(lambda: call("GET", f"/api/v1/exports/precheck?taskId={tid}", admin)) st_pre.add(ms, int((body or {}).get("code", code))) st_pre.done() st_pre.report() if code != 200 or (body or {}).get("code") != 0: print(" 前置检查没通过,跳过创建:%s" % json.dumps(body, ensure_ascii=False)[:200]) return pre = (body or {}).get("data") or {} can = pre.get("canExport") ms, code, body = timed(lambda: call("POST", "/api/v1/exports", admin, { "taskId": tid, "exportName": f"压测导出 {SUFFIX}", "onlyApproved": False, }, timeout=600)) biz = int((body or {}).get("code", code)) msg = (body or {}).get("msg") or "" if can is False: # ⚠️ 前置检查不通过时,创建接口**必须也拒绝** —— 这是契约里那条铁律 # (`canExport=false` 时禁用按钮、不弹警告放行)的服务端一半。 # 把它报成"压测失败"是误导:它会让人以为导出坏了。 blocked = biz == 40035 print(" 前置检查:canExport=false(%d 项 blocking)→ 创建接口%s" % (len(pre.get("blocking") or []), "**正确拒绝**(40035)✓ 铁律在服务端也生效" if blocked else "竟然放行了 ✗(msg=%s)" % msg[:80])) if not blocked: print(" ⚠️ 这是**缺陷**:界面靠 canExport 禁用按钮," "而这里是最后一道服务端防线 —— 它必须自己拦一次") else: st_new = Stat("导出创建(同步打包)") st_new.add(ms, biz) st_new.done() st_new.report() if biz == 0: eid = (body.get("data") or {}).get("id") print(" 导出任务 id=%s(异步打包,这里只量「创建+排队」的耗时)" % eid) # ══════════════ 清理 ══════════════ def cleanup(): # ⚠️ 删库行之前先把**磁盘上的文件路径**取出来:压测会传几百张图、打出几百 MB 的包, # 只删库行的话这些文件会一直躺在 backend/data/uploads 里 # ("删库不删桶"的本地版本 —— 时间一长本地磁盘就是一本糊涂账) disk = [] for tid in TASK_IDS: for stmt in ("select coalesce(std_path, raw_path) from collect_file where task_id=%s;", "select file_path from collect_export_file where task_id=%s;"): for line in sql(stmt % tid).split("\n"): p = line.strip() if p and p != "null": disk.append(p) for tid in TASK_IDS: for tbl in ("collect_validation_issue", "collect_audit_record", "collect_notify_record", "collect_export_file", "collect_file", "collect_submission", "collect_roster", "collect_task_org", "collect_material_def"): sql("delete from %s where task_id=%s;" % (tbl, tid)) sql("delete from collect_task where id=%s;" % tid) for uid in UIDS: sql("delete from collect_message where user_id=%d;" % uid) base = os.path.join(ROOT, "backend", "data", "uploads") removed = 0 for p in disk: full = p if os.path.isabs(p) else os.path.join(base, p) # ⚠️ 只删**确实存在**的本地文件:存储是七牛时这些路径是 bucket key, # 本地根本不存在 —— 那时候要清的是桶,不该在本地乱删(也删不到) if os.path.isfile(full): try: os.remove(full) removed += 1 except OSError: pass for p in (TMP_ROSTER, TMP_IMG): if os.path.exists(p): os.remove(p) print(" 已清理:压测任务 / 名册 / 材料 / 消息 / 临时文件(本地文件 %d 个)" % removed) def main(): # ⚠️ `global` 必须出现在**同一函数里第一次使用该名字之前** —— # 放晚了会报 "name 'X' is used prior to global declaration" global GATE_USERS global BASE ap = argparse.ArgumentParser(description="压测(门禁/上传/读/导出/长跑)") ap.add_argument("--only", default="", help="只跑一个场景:gate / read / upload / export / export-big / soak") ap.add_argument("--conc", type=int, default=0, help="覆盖并发数") ap.add_argument("--clean", action="store_true", help="只清理上次残留") ap.add_argument("--rows", type=int, default=0, help="名册人数(默认 %d)。⚠️ 谈响应时间必须带上规模" % GATE_USERS) ap.add_argument("--base", default="", help="后端地址(默认 %s)。要量上传就另起一个 local 存储的实例" % BASE) ap.add_argument("--with-upload", action="store_true", help="跑上传压测(⚠️ 会往真实对象存储写文件,默认关闭)") ap.add_argument("--img-px", default="2000x1500", help="上传夹具的图片尺寸(真实 JPEG,压缩成本与它成正比)") ap.add_argument("--fixture-only", action="store_true", help="只造夹具并计时(量名册导入耗时用,不做任何压测)") ap.add_argument("--soak", type=int, default=0, help="长时间稳定性:持续 N 分钟,每分钟采一行 RSS/PG 连接数/延迟") ap.add_argument("--rps", type=float, default=5.0, help="长跑强度(请求/秒,默认 5)") ap.add_argument("--files-per-user", type=int, default=1, help="导出大包场景:每人几个材料(默认 1)") ap.add_argument("--split-mb", type=int, default=0, help="导出大包场景:按 N MB 分卷(0 = 不分卷,用服务端默认)") args = ap.parse_args() if args.base: BASE = args.base.rstrip("/") if args.rows > 0: GATE_USERS = args.rows px = tuple(int(x) for x in args.img_px.lower().split("x")) if args.clean: # 残留任务按名字前缀找 for line in sql("select id from collect_task where task_name like '压测任务 %';").split("\n"): if line.strip().isdigit(): TASK_IDS.append(int(line.strip())) cleanup() return 0 print("=" * 100) print(" 压测(M7-09) 后端:%s" % BASE) print(" 规模:名册/门禁 %d 人 读 %d 并发 × %d 轮 上传 %d×%d(需 --with-upload)" % (GATE_USERS, READ_CONC, READ_ROUNDS, UPLOAD_USERS, UPLOAD_PER_USER)) print(" ⚠️ 读接口的结果**必须带规模看**:同一个接口在 60 行与 2000 行上是两件事") print("=" * 100) for acct in ("admin",): os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:login:fail:%s > NUL 2>&1' % acct) _, adm = R.login(BASE, "admin", "Admin@123", "demo") if adm.get("code") != 0: print("管理员登录失败:" + str(adm)) return 2 admin = adm["data"]["token"] # 班主任账号:压测的导出大包场景要真的走一遍"初审 → 终审", # 而这个账号由 scripts/dev-seed-audit.py 摆出来(演示数据的一部分) _, tea = R.login(BASE, TEACHER_USER, TEACHER_PWD, "demo") if tea.get("code") != 0: print("演示班主任 %s 登录失败(先跑 scripts/dev-seed-audit.py):%s" % (TEACHER_USER, tea)) return 2 teacher = tea["data"]["token"] print("\n── 造夹具 ──") t0 = time.perf_counter() tid, tokens = build_fixture(admin, teacher) print(" 任务 %d,名册 %d 人,夹具耗时 %.1fs(含名册导入本身)" % (tid, len(tokens), time.perf_counter() - t0)) if args.fixture_only: # 量"导入 N 行"这件事:接口耗时在上面那条"夹具计时"里,这里补内存与库内行数 print("\n --fixture-only:夹具已造好,直接清理") pid = app_pid(base_port()) print(" 后端 RSS:%.0f MB(pid=%s)" % (rss_mb(pid) or -1, pid)) print(" 库里名册行数:%s(应为 %d)" % (one("select count(*) from collect_roster where task_id=%d and deleted=0;" % tid), GATE_USERS)) cleanup() return 0 stats = {} conc = args.conc try: if args.only in ("", "gate"): print("\n── ① 门禁核验(%d 人同时扫码,每个都要 AES 解密全表比对)──" % GATE_USERS) stats["gate"] = scenario_gate(tid, tokens, conc or GATE_USERS) if args.only in ("", "read"): print("\n── ② 读接口(任务/名册/统计/待办 混合,%d 并发 × %d 轮)──" % (READ_CONC, READ_ROUNDS)) stats["read"] = scenario_read(tid, admin, conc or READ_CONC, READ_ROUNDS) if args.only == "upload" or (args.with_upload and args.only == ""): print("\n── ③ 材料上传(%d 人同时传,压缩是 CPU 密集)──" % UPLOAD_USERS) stats["upload"] = scenario_upload(tid, tokens, conc or UPLOAD_USERS, px) if args.only == "" and not args.with_upload: print("\n── ③ 材料上传(**默认跳过**)──") print(" ⚠️ 当前 dev 的存储是七牛**真实空间**(见 backend/application-local.yml),") print(" 压测上传会打到外部服务并把测试文件灌进真实桶。两种跑法:") print(" ① `--base http://localhost:18080` 打另起的 local 实例(推荐,见 docs/09 §9.5);") print(" ② `XT_STORAGE_PROVIDER=local` 重启当前后端后再加 --with-upload。") if args.only in ("", "export"): print("\n── ④ 导出(前置检查 + 创建)──") scenario_export(tid, admin) if args.only == "export-big": print("\n── ④' 导出大包(造 N 份已通过的单 → 打包 → 验产物)──") scenario_export_big(tid, tokens, admin, teacher, args.files_per_user, args.split_mb) if args.only == "soak" or args.soak > 0: print("\n── ⑤ 长时间稳定性(%d 分钟 × %.1f 请求/秒)──" % (args.soak, args.rps)) scenario_soak(tid, admin, args.soak, args.rps) finally: print("\n── 清理 ──") cleanup() print("\n" + "=" * 100) print(" 结论:看 p95 与**失败分布**。失败码的含义:") print(" 40071 门禁限流(每用户 10 次/小时)—— 说明压测把同一个用户重复用了") print(" 42901 接口限流 50000 内部错(去看 backend/_dev-log.txt) 40301 权限") print("=" * 100) sys.stdout.flush() return 0 if __name__ == "__main__": raise SystemExit(main())