第 16 章 · 插件与扩展
本章目标：掌握 locust-plugins 第三方库的使用、自定义 LoadShape 负载形状、逐步加压阶梯模式，以及如何扩展 Locust 的 Web UI。
16.1 locust-plugins：官方推荐的插件库
locust-plugins 是由 Locust 核心维护者参与的第三方扩展库（当前版本 5.x），提供了大量开箱即用的增强功能：
bash
pip install locust-plugins| 模块 | 功能 |
|---|---|
locust_plugins.users | PlaywrightUser、SeleniumUser、KafkaUser、FtpUser 等 |
locust_plugins.datasets | 数据池（CSV/JSON 驱动） |
locust_plugins.monitors | 监控集成 |
locust_plugins.listeners | Timescale/Grafana 上报 |
使用 PlaywrightUser 做浏览器级压测
python
# locustfile.py
from locust_plugins.users import PlaywrightUser
class BrowserUser(PlaywrightUser):
tasks = [
{
"weight": 1,
"task": lambda page: page.goto("https://example.com"),
"name": "访问首页"
},
{
"weight": 2,
"task": lambda page: (
page.goto("https://example.com/login"),
page.fill("#username", "test"),
page.fill("#password", "pass"),
page.click("button[type=submit]")
),
"name": "登录操作"
}
]为什么用 PlaywrightUser
传统 HttpUser 只压测 HTTP 层，无法覆盖 JavaScript 渲染耗时和前端性能瓶颈。PlaywrightUser 启动真实浏览器执行页面操作，适合评估用户端到端体验。
使用数据池做参数化
python
from locust_plugins.datasets import load_csv_data
class ApiUser(HttpUser):
# 从 CSV 加载测试数据，循环使用
dataset = load_csv_data("test_users.csv")
@task
def login(self):
credentials = next(self.dataset)
self.client.post("/api/login", json={
"username": credentials["username"],
"password": credentials["password"]
})16.2 自定义负载形状 LoadShape
默认情况下，Locust 通过 Web UI 或命令行参数设定固定的并发用户数。但真实压测场景往往需要动态调整负载曲线——比如阶梯加压、峰值脉冲或昼夜模拟。
Locust 提供了 LoadTestShape 抽象基类，通过重写 tick() 方法来控制每一时刻的用户数：
python
from locust import HttpUser, task, between, LoadTestShape
# 负载形状类——独立于 User 类定义
class StagedLoadShape(LoadTestShape):
"""分阶段负载：每阶段持续指定时间后切换到下一档"""
stages = [
{"duration": 60, "users": 20, "spawn_rate": 5}, # 第 1 分钟: 20 用户
{"duration": 120, "users": 50, "spawn_rate": 10}, # 第 2 分钟: 50 用户
{"duration": 180, "users": 100, "spawn_rate": 20}, # 第 3 分钟: 100 用户
{"duration": 240, "users": 200, "spawn_rate": 30}, # 第 4 分钟: 200 用户
{"duration": 300, "users": 50, "spawn_rate": 40}, # 第 5 分钟: 回落到 50
]
def tick(self):
"""返回 (user_count, spawn_rate)，返回 None 则终止测试"""
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 ApiUser(HttpUser):
wait_time = between(1, 3)
@task
def query(self):
self.client.get("/api/data")运行时无需指定 -u 和 -r 参数，Locust 会自动按 tick() 返回值调度：
bash
locust -f locustfile.py --headless --host https://staging.example.com关键要点
tick()每秒调用一次；- 返回值是元组
(目标用户数, 孵化速率)； - 返回
None表示结束整个测试； self.get_run_time()返回自测试开始以来的秒数。
注意事项
一个脚本中只能定义一个 LoadTestShape 子类。如果定义了多个，Locust 会抛出异常。将 Shape 与 User 类写在同一个文件中即可生效。
16.3 实战：逐步加压找拐点
性能测试的核心目标之一是找到系统的容量拐点（即响应时间开始急剧上升的并发阈值）。以下是一个精细的阶梯加压示例：
python
from locust import LoadTestShape
class RampUpShape(LoadTestShape):
"""
每 30 秒增加 25 个用户，从 25 到 500，
观察哪个区间响应时间突破 SLA 阈值（如 p95 > 500ms）
"""
step_interval = 30 # 每步持续时间
step_users = 25 # 每步增加用户数
max_users = 500 # 最大用户数
def tick(self):
run_time = self.get_run_time()
current_step = int(run_time // self.step_interval)
target_users = min(
(current_step + 1) * self.step_users,
self.max_users
)
if target_users > self.max_users:
return None # 达到上限，结束测试
return (target_users, 10) # 孵化速率固定为 10/s配合第 19 章的监控方案，你可以在 Grafana 图表上直观看到"哪一档用户数导致 p95 响应时间飙升"，从而精确定位系统容量上限。
16.4 扩展 Web UI
Locust 的 Web UI 基于 Flask 构建，可以通过 environment.web_ui.app 访问底层 Flask 实例并添加自定义路由或模板：
python
from locust import events
@events.init.add_listener
def on_locust_init(environment, **kwargs):
if environment.web_ui:
# 注册自定义 REST API 端点
@environment.web_ui.app.route("/custom-stats")
def custom_stats():
"""暴露自定义统计数据给外部监控"""
stats = environment.stats
total = stats.total
return {
"current_rps": round(total.current_rps, 2),
"fail_ratio": round(total.fail_ratio, 4),
"p95_response_time": total.get_response_time_percentile(0.95),
"active_users": environment.runner.user_count,
}
# 注入自定义 HTML 到 Web UI
@environment.web_ui.app.after_request
def add_header(response):
response.headers["X-Custom-Header"] = "locust-extended"
return response还可以通过 events.init 向 Web UI 页面注入自定义 JavaScript/CSS，实现品牌定制或额外图表展示。
扩展事件钩子实现自定义报告
python
import time
from locust import events
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
print(f"[{time.strftime('%H:%M:%S')}] 测试开始")
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
"""每个请求完成后触发——可推送指标到外部系统"""
if exception:
print(f"[FAIL] {request_type} {name}: {exception}")
elif response_time > 1000: # 记录慢请求
print(f"[SLOW] {request_type} {name}: {response_time:.0f}ms")
@events.quitting.add_listener
def on_quitting(environment, **kwargs):
"""测试结束时检查失败率是否超标"""
fail_ratio = environment.stats.total.fail_ratio
if fail_ratio > 0.05:
print(f"\n❌ 失败率 {fail_ratio:.1%} 超过 5% 阈值！")
environment.process_exit_code = 1 # 让 CI 判定为失败
else:
print(f"\n✅ 失败率 {fail_ratio:.1%} 在允许范围内")本章小结
locust-plugins提供 PlaywrightUser、数据池等高级能力；LoadTestShape.tick()控制动态负载曲线，返回None终止测试；- 阶梯加压是定位系统容量拐点的标准方法；
- 通过
events.init可扩展 Web UI 和注册自定义端点； events.quitting中设置process_exit_code = 1可让 CI 正确判定压测失败。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. LoadTestShape 的 tick() 方法返回什么值表示终止测试？
2. 关于 locust-plugins 库，以下说法正确的是？
3. LoadTestShape 的 tick() 方法每隔多长时间被调用一次？
4. 如何让 CI 系统在压测失败率超标时正确判定构建失败？
🛠️ 动手实践
- 编写一个包含 3 个阶段的
StagedLoadShape：低负载(10 用户×2 分钟) → 高负载(100 用户×3 分钟) → 冷却(5 用户×1 分钟)，并在 headless 模式下运行验证。 - 安装
locust-plugins，用load_csv_data实现 10 组不同账号的登录参数化压测。 - 通过
events.init为你的 Web UI 添加/p99自定义端点，返回当前 p99 响应时间。
完成练习后，进入下一章：性能调优与请求率提升。