第 8 章 · 事件钩子与自定义统计
本章目标：掌握 Locust 的事件系统（events.request/test_start/test_stop 等），学会注入自定义指标和监听关键节点。
8.1 事件系统概览
Locust 的事件系统基于 locust.events 模块，你可以在测试的不同阶段注册回调函数：
python
from locust import events
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
print("压测开始！")
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
print("压测结束！")常用事件：
| 事件 | 触发时机 |
|---|---|
test_start | 整个压测开始时（一次） |
test_stop | 压测结束时（一次） |
request | 每次 HTTP 请求完成后 |
worker_report | Worker 向 Master 汇报时 |
spawning_complete | 所有虚拟用户已启动 |
8.2 监听 request 事件
每次 HTTP 请求完成都会触发 request 事件，可用于自定义日志、告警或统计：
python
from locust import events
import time
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, context, **kwargs):
if exception:
print(f"❌ {request_type} {name} 失败: {exception}")
elif response_time > 2000: # 超过 2 秒的慢请求
print(f"⚠️ 慢请求 {name}: {response_time:.0f}ms")8.3 自定义指标上报
通过 events.request.fire() 可以将非 HTTP 操作纳入 Locust 统计体系：
python
from locust import HttpUser, task, between, events
class CustomMetricsUser(HttpUser):
wait_time = between(1, 2)
@task
def process_data(self):
start = time.time()
# 执行一些非 HTTP 操作（如数据库查询）
result = self.run_query()
elapsed_ms = (time.time() - start) * 1000
# 上报为自定义请求，纳入 Locust 的响应时间图表
events.request.fire(
request_type="DB",
name="SELECT users",
response_time=elapsed_ms,
response_length=len(str(result)),
exception=None,
)8.4 在 Web UI 中展示自定义数据
python
from locust import events
from locust.web import app
from flask import jsonify
custom_stats = {"errors": 0}
@events.request.add_listener
def track_errors(exception, **kwargs):
if exception:
custom_stats["errors"] += 1
@app.route("/custom-stats")
def custom_stats_endpoint():
return jsonify(custom_stats)访问 http://localhost:8089/custom-stats 即可获取实时数据。
本章小结
events模块提供 test_start/test_stop/request 等钩子；- 通过
events.request.fire()可以上报非 HTTP 操作的自定义指标； - 结合 Flask 路由可以在 Web UI 中暴露自定义监控端点。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 哪个事件在整个压测开始时只触发一次？
2. 如何让一段非 HTTP 逻辑出现在 Locust 的响应时间统计中？
3. request 事件回调中哪个参数表示请求失败时的异常信息？
4. Locust 的 Web UI 是基于哪个框架构建的？
🛠️ 动手实践
- 编写一个 request 事件监听器，当响应时间超过阈值时打印警告。
- 用 events.request.fire() 将一段纯 Python 计算逻辑的上报为 "CALC" 类型的自定义指标。
- 在 Web UI 中添加一个
/health端点返回当前活跃虚拟用户数。