diff --git a/tests/performance/README.md b/tests/performance/README.md new file mode 100644 index 00000000..5023cde6 --- /dev/null +++ b/tests/performance/README.md @@ -0,0 +1,32 @@ +# 性能压测方案（Issue #94） + +## 范围 + +1. **REST API 并发**：Locust 梯度 10/50/100/500/1000 并发，覆盖 + 登录 -> 设备查询 -> 数据上报 -> 报表查询完整路径（`locustfile.py`） +2. **数据量**：百万级抄表/营收记录下的分页查询与报表聚合（结合 `db/` 种子数据， + 关注慢查询日志 > 500ms 的语句并补索引） +3. **WebSocket 长连接**：`websocket_stress.py` 支持 100-5000 并发连接， + 输出建连成功率与 P50/P95/P99 往返延迟 + +## 运行 + +```bash +pip install locust websockets +mkdir -p reports +# REST API（示例：100 并发 5 分钟） +locust -f tests/performance/locustfile.py --host http://127.0.0.1:8000 \ + -u 100 --spawn-rate 10 --run-time 5m --headless --html reports/api_100.html +# WebSocket（示例：500 连接 60 秒） +python tests/performance/websocket_stress.py --url ws://127.0.0.1:8000/ws \ + --concurrency 500 --duration 60 +``` + +## 验收基线（测试环境 2C4G） + +| 指标 | 目标 | +|------|------| +| API P95 延迟（100 并发） | ≤ 300ms | +| API 错误率（500 并发） | ≤ 0.5% | +| WebSocket 500 连接建连成功率 | ≥ 99% | +| WS 消息 P95 往返 | ≤ 200ms | diff --git a/tests/performance/__init__.py b/tests/performance/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/performance/__pycache__/locustfile.cpython-312.pyc b/tests/performance/__pycache__/locustfile.cpython-312.pyc new file mode 100644 index 00000000..53fb3825 Binary files /dev/null and b/tests/performance/__pycache__/locustfile.cpython-312.pyc differ diff --git a/tests/performance/__pycache__/websocket_stress.cpython-312.pyc b/tests/performance/__pycache__/websocket_stress.cpython-312.pyc new file mode 100644 index 00000000..b982474f Binary files /dev/null and b/tests/performance/__pycache__/websocket_stress.cpython-312.pyc differ diff --git a/tests/performance/locustfile.py b/tests/performance/locustfile.py new file mode 100644 index 00000000..619c3a6e --- /dev/null +++ b/tests/performance/locustfile.py @@ -0,0 +1,59 @@ +"""REST API 压测脚本（Issue #94）- Locust + +覆盖完整用户路径：登录 -> 设备查询 -> 数据上报 -> 报表查询 +并发梯度：locust -u 10/50/100/500/1000 --spawn-rate 10 + +运行： + pip install locust + locust -f tests/performance/locustfile.py --host http://127.0.0.1:8000 \ + -u 100 --spawn-rate 10 --run-time 5m --headless \ + --html reports/locust_$(date +%F_%H%M).html +""" + +from locust import HttpUser, between, task + + +class WaterManagementUser(HttpUser): + wait_time = between(0.5, 2.0) + token = None + + def on_start(self): + """登录获取 token（若接口不存在则以匿名继续，保证压测可运行）""" + resp = self.client.post("/api/auth/login", + json={"username": "perf", "password": "perf123"}, + catch_response=True) + if resp.status_code == 200: + try: + self.token = resp.json().get("data", {}).get("access_token") + except Exception: + self.token = None + else: + resp.success() # 登录接口未实现时不计入失败 + + @property + def headers(self): + return {"Authorization": f"Bearer {self.token}"} if self.token else {} + + @task(5) + def list_devices(self): + self.client.get("/api/devices?page=1&page_size=20", headers=self.headers, + name="/api/devices") + + @task(3) + def device_detail(self): + self.client.get("/api/devices/1", headers=self.headers, name="/api/devices/{id}") + + @task(2) + def report_data(self): + self.client.post("/api/data/report", + json={"device_id": 1, "metric": "flow", "value": 12.5}, + headers=self.headers, name="/api/data/report") + + @task(2) + def billing_report(self): + self.client.get("/api/billing/report?month=2026-07", headers=self.headers, + name="/api/billing/report") + + @task(1) + def health(self): + self.client.get("/health", name="/health") diff --git a/tests/performance/websocket_stress.py b/tests/performance/websocket_stress.py new file mode 100644 index 00000000..f8deb84f --- /dev/null +++ b/tests/performance/websocket_stress.py @@ -0,0 +1,83 @@ +"""WebSocket 长连接并发压测（Issue #94） + +支持 100 - 5000 并发连接梯度，统计建连成功率、消息往返延迟（P50/P95/P99）。 + +运行： + python tests/performance/websocket_stress.py --url ws://127.0.0.1:8000/ws \ + --concurrency 500 --duration 60 +""" + +from __future__ import annotations + +import argparse +import asyncio +import statistics +import time + +try: + import websockets +except ImportError: # pragma: no cover + websockets = None + + +class Stats: + def __init__(self) -> None: + self.connected = 0 + self.failed = 0 + self.latencies: list[float] = [] + + def summary(self) -> str: + if not self.latencies: + lat = "无消息样本" + else: + ls = sorted(self.latencies) + p = lambda q: ls[min(len(ls) - 1, int(q * len(ls)))] + lat = (f"P50={p(0.5)*1000:.1f}ms P95={p(0.95)*1000:.1f}ms " + f"P99={p(0.99)*1000:.1f}ms avg={statistics.mean(ls)*1000:.1f}ms") + return (f"建连成功={self.connected} 失败={self.failed} | 往返延迟 {lat}") + + +async def worker(url: str, duration: float, stats: Stats) -> None: + try: + async with websockets.connect(url, open_timeout=10) as ws: + stats.connected += 1 + end = time.time() + duration + while time.time() < end: + t0 = time.perf_counter() + await ws.send('{"type":"ping"}') + try: + await asyncio.wait_for(ws.recv(), timeout=5) + stats.latencies.append(time.perf_counter() - t0) + except asyncio.TimeoutError: + pass + await asyncio.sleep(1) + except Exception: + stats.failed += 1 + + +async def main_async(url: str, concurrency: int, duration: float, ramp: float) -> None: + stats = Stats() + tasks = [] + for i in range(concurrency): + tasks.append(asyncio.create_task(worker(url, duration, stats))) + await asyncio.sleep(ramp) # 平滑加压，避免瞬时 SYN 风暴 + await asyncio.gather(*tasks) + print(stats.summary()) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--url", default="ws://127.0.0.1:8000/ws") + ap.add_argument("--concurrency", type=int, default=100, help="并发连接数(100-5000)") + ap.add_argument("--duration", type=float, default=60, help="每连接压测时长(秒)") + ap.add_argument("--ramp", type=float, default=0.01, help="建连间隔(秒)") + args = ap.parse_args() + if websockets is None: + raise SystemExit("请先 pip install websockets") + if not 1 <= args.concurrency <= 5000: + raise SystemExit("concurrency 需在 1-5000 之间") + asyncio.run(main_async(args.url, args.concurrency, args.duration, args.ramp)) + + +if __name__ == "__main__": + main()