第 20 章 · 综合实战：全链路压测

本章目标：将前 19 章的知识串联起来，对一个 FastAPI 应用执行完整的性能测试流程——从编写脚本到分布式执行，再到分析报告和提出优化建议。

20.1 被测系统：一个简单的任务管理 API

假设我们有一个部署在 staging 环境的 FastAPI 应用：

python

# app.py — 被测的 FastAPI 应用（staging 环境）
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
import time


app = FastAPI()
db = {"tasks": [], "next_id": 1}  # 模拟数据库


class TaskCreate(BaseModel):
    title: str
    priority: int = 0


@app.get("/api/tasks")
async def list_tasks():
    return db["tasks"]


@app.post("/api/tasks")
async def create_task(task: TaskCreate):
    time.sleep(0.05)  # 模拟 DB 写入延迟
    item = {"id": db["next_id"], **task.model_dump()}
    db["next_id"] += 1
    db["tasks"].append(item)
    return item


@app.get("/api/tasks/{task_id}")
async def get_task(task_id: int):
    for t in db["tasks"]:
        if t["id"] == task_id:
            return t
    raise HTTPException(status_code=404)

20.2 第一步：编写 Locust 压测脚本

按照真实用户的使用比例设计任务权重——读多写少是最常见的模式：

python

# locustfile.py
import random
from locust import HttpUser, task, between, LoadTestShape


class StagedShape(LoadTestShape):
    """阶梯加压：25 → 50 → 100 → 200 用户"""
    stages = [
        {"duration": 60,  "users": 25,  "spawn_rate": 5},
        {"duration": 120, "users": 50,  "spawn_rate": 10},
        {"duration": 180, "users": 100, "spawn_rate": 15},
        {"duration": 240, "users": 200, "spawn_rate": 20},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                return (stage["users"], stage["spawn_rate"])
        return None


class TaskApiUser(HttpUser):
    wait_time = between(0.5, 2)
    host = "https://staging.example.com"

    def on_start(self):
        """初始化：创建一些初始任务供后续读取"""
        self.task_ids = []
        for i in range(3):
            resp = self.client.post("/api/tasks", json={
                "title": f"Setup task {i}", "priority": 0
            })
            if resp.status_code == 200:
                self.task_ids.append(resp.json()["id"])

    @task(8)  # 权重 8 — 读操作占 80%
    def list_tasks(self):
        self.client.get("/api/tasks", name="/api/tasks [list]")

    @task(4)  # 权重 4 — 读单个任务占 40%
    def get_task(self):
        if self.task_ids:
            tid = random.choice(self.task_ids)
            self.client.get(f"/api/tasks/{tid}", name="/api/tasks [detail]")

    @task(2)  # 权重 2 — 写操作占 20%
    def create_task(self):
        self.client.post("/api/tasks", json={
            "title": f"Load test task {random.randint(1, 10000)}",
            "priority": random.randint(0, 3)
        }, name="/api/tasks [create]")

使用 name 分组统计

name="/api/tasks [list]" 把不同参数的请求归入同一个统计桶。如果不设置 name，每个不同的 URL 会单独统计，导致报表混乱。

20.3 第二步：本地验证

在正式压测前先做小规模验证：

bash

# 1. 验证脚本语法无误
python -c "import locustfile" || echo "语法错误"

# 2. 小规模 dry run（5 用户 × 30 秒）
locust -f locustfile.py --headless \
  --users 5 --spawn-rate 5 --run-time 30s \
  --host https://staging.example.com \
  --only-summary

检查输出：

✅ 所有接口的失败率为 0%
✅ 响应时间在合理范围内（< 500ms）
✅ RPS 与预期的用户数和 wait_time 匹配

20.4 第三步：分布式执行

确认脚本正常后，启动完整压测：

bash

# 单机多进程模式（8 核 CPU 的压力机）
locust -f locustfile.py \
  --processes 8 \
  --headless \
  --host https://staging.example.com \
  --csv results/full_test \
  --html results/full_report.html

# 或跨机器分布式模式
# 主控机：
locust -f locustfile.py --master --headless \
  --csv results/full_test --html results/full_report.html
# 各压力机：
locust -f locustfile.py --worker --master-host=<master-ip>

20.5 第四步：分析报告

测试结束后，打开 full_report.html 和 CSV 文件进行分析：

分析清单

检查项	通过标准	本例结果
p95 @ 25 users	< 200ms	✅ 145ms
p95 @ 50 users	< 300ms	✅ 280ms
p95 @ 100 users	< 500ms	⚠️ 520ms
p95 @ 200 users	< 800ms	❌ 2100ms
失败率	< 1%	✅ 0.2%
RPS 峰值	> 500/s	✅ 850/s

结论与优化建议

text

📊 性能测试结论
━━━━━━━━━━━━━━━━━━━━━━━
系统容量拐点：约 80 并发用户
最大安全吞吐量：~600 RPS
瓶颈位置：POST /api/tasks 接口（同步 time.sleep 模拟 DB 写入）

优化建议（按优先级排序）：
1. 将 time.sleep(0.05) 替换为异步数据库操作
   → 预期提升 3-5 倍吞吐量
2. 增加 uvicorn workers 数量（当前 1 → 建议 4）
   → 充分利用多核 CPU
3. 引入连接池复用数据库连接
   → 减少 TCP 握手开销
4. 对 GET /api/tasks 列表添加分页和缓存
   → 降低数据传输量和重复计算

20.6 第五步：验证优化效果

实施优化后，用完全相同的负载形状重新压测，对比前后数据：

指标	优化前	优化后	提升
p95 @ 200 users	2100ms	380ms	5.5×
最大 RPS	850/s	3200/s	3.8×
容量拐点	~80 用户	~350 用户	4.4×

性能测试的核心价值

不是生成一张漂亮的报告，而是通过可复现的数据驱动决策：知道系统能扛多少、哪里是短板、优化后到底提升了多少。

本章小结

全链路压测流程：编写脚本 → 本地验证 → 分布式执行 → 分析报告 → 优化建议；
用 @task(weight) 模拟真实的读/写比例，用 name= 参数分组统计；
阶梯加压找到容量拐点后记录关键指标作为基线；
优化后必须用相同负载重新测试才能公平对比。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 在 Locust 中使用 name 参数的主要目的是什么？

2. 压测报告中发现"增加并发用户数但 RPS 不再增长"，最合理的结论是？

3. 为什么优化后必须使用完全相同的负载形状重新压测？

4. 以下哪项不属于压测前的本地验证步骤？

🛠️ 动手实践

为你自己开发的一个 API 服务编写完整的 Locust 压测方案（含 Shape、参数化和阈值判定），并跑出一份 HTML 报告。
在你的 FastAPI 应用中故意引入一个 N+1 查询问题，然后用本章的方法论定位它并修复，记录修复前后的性能对比。
尝试用 Docker Compose 编排 Locust master + 3 worker + Prometheus + Grafana，实现一键启动的全链路压测监控环境。

🎉 恭喜完成全部 20 章！建议接下来学习Agent 工程课程了解如何用 AI 辅助编写压测脚本，或者探索 Playwright 进行浏览器级别的端到端测试。

第 20 章 · 综合实战：全链路压测 ​

20.1 被测系统：一个简单的任务管理 API ​

20.2 第一步：编写 Locust 压测脚本 ​

20.3 第二步：本地验证 ​

20.4 第三步：分布式执行 ​

20.5 第四步：分析报告 ​

分析清单 ​

结论与优化建议 ​

20.6 第五步：验证优化效果 ​

本章小结 ​