# -*- 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__))) BASE = "http://localhost:8080" 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 build_fixture(admin): """造一个任务 + 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", }) 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": 2048, "maxCount": 3, "minSizeKb": 1, "minLongSide": 100, }) call("PUT", f"/api/v1/tasks/{tid}/orgs", admin, {"orgs": [{"orgId": org["id"], "expectedCount": GATE_USERS}]}) # 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)) build_roster(rows) body = R.upload_roster(BASE, admin, tid, TMP_ROSTER, extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]}) # ⚠️ **判据是 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))) call("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): """造一张尽量接近真实照片的图（纯色会被压得很小，失真）。""" import random random.seed(7) # 最小合法 JPEG 头 + 随机填充（不是真 JPEG，但上传接口只看扩展名与大小） 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") 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): """上传：并发上传图片（压缩是 CPU 密集）。""" st = Stat("材料上传") make_img() sub_ids = {} # 每个用户先建草稿 for t in tokens[:conc]: 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"] 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 = [] for t in tokens[:conc]: 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 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(): for tid in TASK_IDS: for tbl in ("collect_validation_issue", "collect_audit_record", "collect_notify_record", "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) for p in (TMP_ROSTER, TMP_IMG): if os.path.exists(p): os.remove(p) print(" 已清理：压测任务 / 名册 / 材料 / 消息 / 临时文件") def main(): # ⚠️ `global` 必须出现在**同一函数里第一次使用该名字之前** —— # 放晚了会报 "name 'X' is used prior to global declaration" global GATE_USERS ap = argparse.ArgumentParser(description="压测（门禁/上传/读/导出）") ap.add_argument("--only", default="", help="只跑一个场景：gate / read / upload / export") 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("--with-upload", action="store_true", help="跑上传压测（⚠️ 会往真实对象存储写文件，默认关闭）") args = ap.parse_args() if args.rows > 0: GATE_USERS = args.rows 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"] print("\n── 造夹具 ──") t0 = time.perf_counter() tid, tokens = build_fixture(admin) print(" 任务 %d，名册 %d 人，夹具耗时 %.1fs" % (tid, len(tokens), time.perf_counter() - t0)) 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) if args.only == "" and not args.with_upload: print("\n── ③ 材料上传（**默认跳过**）──") print(" ⚠️ 当前 dev 的存储是七牛**真实空间**（见 backend/application-local.yml），") print(" 压测上传会打到外部服务并把测试文件灌进真实桶。要跑就加 --with-upload，") print(" 且最好先把存储切成本地：XT_STORAGE_PROVIDER=local 重启后端。") if args.only in ("", "export"): print("\n── ④ 导出（前置检查 + 创建）──") scenario_export(tid, admin) 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())