一、为什么Locust自带的Web界面不够用
用Locust做压力测试的同学应该都有过这样的体验：测试跑完之后打开Web界面，页面上只显示了请求次数、响应时间、失败率这些基础指标。但是，你真正关心的业务指标呢？比如用户下单成功了几笔、库存扣减耗时多少、第三方接口回调成功率如何——这些统统看不到。
这不是你用法不对，而是Locust的设计定位决定的。它本质上是一个通用的HTTP压力测试工具，自带的Web界面展示的是通用的网络层指标。当你想追踪业务层的自定义数据时，就必须自己动手扩展。
这篇文章就来手把手教大家，怎么在Locust里面加上自定义指标的采集，并且让这些数据能直接在Web界面上展示出来。整个过程不需要改动Locust的源码，只需要写几行代码就能搞定。
二、深入理解Locust的指标采集机制
2.1 Locust的指标体系是怎么回事
要搞明白怎么加自定义指标，首先得搞清楚Locust现有的指标是怎么工作的。简单说一下，Locust内部维护着一套统计系统，每次发请求都会往里面写数据，Web界面定期从这套系统里拉数据展示。
Locust提供了一个叫做events的机制，相当于一个事件总线。在测试运行过程中，各种事件会往这个总线上发，任何模块都可以订阅这些事件。这就给了我们一个口子——我们可以在这个总线上塞进自己的自定义事件。
2.2 事件总线的工作原理
事件总线就像是快递站的包裹分拣系统。发件人把包裹（事件）扔进去，标注好类型，分拣系统就会把对应类型的包裹送到对应的收件人（处理器）手里。
在Locust里，常见的事件类型包括：REQUEST（请求事件）、LOGIN（登录事件）、CUSTOM（自定义事件）等。我们要用的就是CUSTOM这个类型，它是专门留给用户扩展的。
技术栈：Python
# 技术栈：Python
# 文件：custom_events_demo.py
# 演示Locust事件总线的基本用法
from locust import events, task, FastHttpUser
from time import sleep
import random
# 定义一个事件处理函数，监听Locust的CUSTOM事件
# 每次有自定义事件发出时，这个函数就会被调用
@events.custom.add_listener
def handle_custom_event(environment, **kwargs):
"""
处理自定义事件
:param environment: Locust的运行环境对象
:param kwargs: 事件中携带的所有数据，键值对形式
"""
# 打印事件名称和携带的数据，方便调试
print(f"[自定义事件捕获] 名称: {kwargs.get('name')}, 值: {kwargs.get('value')}, 额外数据: {kwargs.get('extra')}")
# 定义用户行为类
class MyUser(FastHttpUser):
"""模拟用户的测试类"""
# 等待时间设置：最小5秒，最大15秒
wait_time = lambda self: random.uniform(5, 15)
@task
def my_task(self):
"""
测试任务：模拟一个业务流程
"""
# 模拟发送一个请求
start_time = time.time()
self.client.get("/api/orders")
elapsed = time.time() - start_time
# 发出一个自定义事件，告诉Locust这次业务操作的结果
# 注意：发事件要用 events.custom.fire()
events.custom.fire(
name="业务处理耗时", # 自定义指标名称
value=elapsed, # 指标数值
unit="秒", # 单位
extra={"order_id": "12345"} # 额外信息（可选）
)
2.3 指标数据在内部是怎么存储的
Locust内部有一个StatsEntry类，每个指标都对应一个StatsEntry实例。这个实例里记录了请求次数、平均响应时间、最大响应时间、最小响应时间、失败次数等等。
这些StatsEntry实例被存放在一个叫做RequestStats的对象里面，可以理解为一张大的统计表。Web界面就是通过访问这张表来获取数据的。
技术栈：Python
# 技术栈：Python
# 文件：stats_inspection.py
# 演示如何查看Locust内部的统计数据
from locust.stats import RequestStats, StatsEntry
import time
# 创建一个新的统计数据对象
stats = RequestStats()
# 向统计表里添加一条记录，模拟一次请求
# total_count: 总请求数
# num_failures: 失败次数
# total_response_time: 总响应时间（毫秒）
# 返回值是当前的StatsEntry对象
stats_entry = stats.log_request(
method="GET", # HTTP方法
name="/api/orders", # 接口名称
response_time=125.3, # 本次响应时间（毫秒）
content_length=512, # 响应体大小（字节）
error=None # 错误信息，None表示成功
)
# 获取统计对象的摘要信息
# request_stats属性包含了按请求名分组的所有StatsEntry
print("统计摘要信息：")
print(f"总请求数: {stats.total.request_count}") # 总请求次数
print(f"平均响应时间: {stats.total.avg_response_time}") # 平均响应时间
print(f"失败请求数: {stats.total.num_failures}") # 失败次数
print(f"请求总数: {stats.total.num_requests}") # 请求数（包括成功和失败）
三、自定义指标采集的完整实现方案
3.1 创建自定义统计类
要采集自定义指标，最干净的做法是自己写一个统计类，继承Locust的RequestStats，然后往里面加新字段。不过考虑到兼容性和维护成本，我们这里采用一个更简洁的方案：利用事件总线加一个内存计数器。
这个方案的好处是简单、不容易出错，坏处是你得自己管数据的生命周期。对于大多数场景来说，足够了。
技术栈：Python
# 技术栈：Python
# 文件：custom_stats_collector.py
# 自定义指标采集器的完整实现
import time
import threading
from collections import defaultdict
from locust import events
class CustomStatsCollector:
"""
自定义指标采集器
负责收集和统计Locust测试过程中产生的业务级自定义指标
"""
def __init__(self):
# 线程锁，保证多线程环境下数据读写安全
self._lock = threading.Lock()
# 用字典存储所有指标的统计数据
# 结构: { 指标名称: { "total": 总数, "sum": 值之和, "max": 最大值, "min": 最小值, "min_time": 首次记录时间 } }
self._data = defaultdict(lambda: {
"total": 0, # 记录条数
"sum": 0.0, # 所有值的总和
"max": 0.0, # 最大值
"min": float("inf"), # 最小值
"timestamps": [] # 时间戳列表，用于计算趋势
})
# 注册事件监听器，自动接收发来的自定义事件
events.custom.add_listener(self._on_custom_event)
def _on_custom_event(self, environment, **kwargs):
"""
事件回调函数：每次收到自定义事件时自动调用
"""
name = kwargs.get("name", "unnamed") # 指标名称
value = kwargs.get("value", 0) # 指标值
extra = kwargs.get("extra", {}) # 额外数据
with self._lock:
# 获取或创建该指标的统计记录
entry = self._data[name]
entry["total"] += 1 # 计数加1
entry["sum"] += float(value) # 累加总值
entry["max"] = max(entry["max"], float(value)) # 更新最大值
entry["min"] = min(entry["min"], float(value)) # 更新最小值
entry["timestamps"].append(time.time()) # 记录时间戳
# 如果时间戳太多，只保留最近1000条，防止内存爆炸
if len(entry["timestamps"]) > 1000:
entry["timestamps"] = entry["timestamps"][-1000:]
def get_stats(self):
"""
获取所有自定义指标的统计汇总
:return: 一个字典，包含每个指标的详细统计数据
"""
result = {}
with self._lock:
for name, data in self._data.items():
result[name] = {
"count": data["total"],
"avg": data["sum"] / data["total"] if data["total"] > 0 else 0,
"max": data["max"],
"min": data["min"] if data["min"] != float("inf") else 0,
"total_sum": data["sum"]
}
return result
def reset(self):
"""清空所有统计数据，开始新一轮统计"""
with self._lock:
self._data.clear()
# 全局实例，测试脚本中直接引用这个对象
custom_stats = CustomStatsCollector()
3.2 在测试脚本中使用自定义指标
采集器建好了，接下来就是在测试脚本里用它。每次你的业务操作完成后，发一个事件就行。
技术栈：Python
# 技术栈：Python
# 文件：locustfile.py
# 完整的Locust测试脚本，包含自定义指标采集
import random
from time import sleep
from locust import FastHttpUser, task, between, events
from custom_stats_collector import custom_stats # 导入自定义采集器
class ECommerceUser(FastHttpUser):
"""
电商场景压力测试用户
模拟用户浏览商品、加入购物车、下单的完整流程
"""
# 用户请求间隔时间：5秒到20秒之间随机
wait_time = between(5, 20)
@task(3) # 权重3，表示这个任务被选中的概率是其他的3倍
def browse_products(self):
"""浏览商品列表，这个任务执行频率较高"""
# 记录开始时间，用来算接口耗时
start = time.time()
response = self.client.get(
"/api/products",
params={"page": random.randint(1, 10), "size": 20},
catch_response=True # 开启响应捕获，可以手动判定成功或失败
)
# 计算接口耗时（毫秒）
api_cost = (time.time() - start) * 1000
# 发送自定义指标事件：记录商品列表接口的响应时间
events.custom.fire(
name="商品列表接口耗时",
value=round(api_cost, 2),
unit="ms"
)
# 检查HTTP状态码
if response.status_code != 200:
response.failure("商品列表接口返回异常状态码: %d" % response.status_code)
# 如果商品列表有数据，模拟点击进入详情页
if len(response.json().get("data", [])) > 0:
product_id = response.json()["data"][0]["id"]
self._view_product_detail(product_id)
@task(1) # 权重1，低频任务
def add_to_cart(self):
"""加入购物车操作"""
start = time.time()
product_id = random.randint(1, 1000)
quantity = random.randint(1, 5)
response = self.client.post(
"/api/cart",
json={"product_id": product_id, "quantity": quantity},
catch_response=True
)
# 发送自定义指标：购物车操作耗时
cart_cost = (time.time() - start) * 1000
events.custom.fire(
name="购物车操作耗时",
value=round(cart_cost, 2),
unit="ms"
)
# 手动判定成功或失败
if response.status_code == 200 and response.json().get("success"):
response.success() # 标记为成功
else:
response.failure("加入购物车失败: %s" % response.text[:200])
@task(1) # 权重1，低频任务
def place_order(self):
"""
下单操作——核心业务，需要采集更多细粒度的指标
"""
start_total = time.time()
# 第一步：校验库存
inventory_start = time.time()
self.client.get("/api/inventory/check", params={"sku": "SKU-2024-001"})
inventory_cost = (time.time() - inventory_start) * 1000
events.custom.fire(
name="库存校验耗时",
value=round(inventory_cost, 2),
unit="ms"
)
# 第二步：创建订单
order_start = time.time()
response = self.client.post(
"/api/orders",
json={
"items": [{"product_id": 101, "quantity": 2}],
"shipping_address": "北京市朝阳区xxx路xxx号"
},
catch_response=True
)
order_cost = (time.time() - order_start) * 1000
# 发送自定义指标：订单创建耗时
events.custom.fire(
name="订单创建耗时",
value=round(order_cost, 2),
unit="ms"
)
# 第三步：支付处理
if response.status_code == 200:
order_id = response.json().get("order_id")
payment_start = time.time()
self.client.post("/api/payment", json={"order_id": order_id})
payment_cost = (time.time() - payment_start) * 1000
events.custom.fire(
name="支付处理耗时",
value=round(payment_cost, 2),
unit="ms"
)
# 第四步：记录整个下单流程的总耗时
total_cost = (time.time() - start_total) * 1000
events.custom.fire(
name="下单流程总耗时",
value=round(total_cost, 2),
unit="ms"
)
if response.status_code != 200:
response.failure("下单失败")
def _view_product_detail(self, product_id):
"""查看商品详情页（内部方法）"""
start = time.time()
response = self.client.get(
f"/api/products/{product_id}",
catch_response=True
)
detail_cost = (time.time() - start) * 1000
events.custom.fire(
name="商品详情接口耗时",
value=round(detail_cost, 2),
unit="ms"
)
if response.status_code != 200:
response.failure("商品详情接口异常")
# 测试结束时，打印自定义统计摘要
@events.test_stop.add_listener
def print_custom_summary(environment, **kwargs):
"""测试结束钩子：打印所有自定义指标的汇总数据"""
print("\n" + "=" * 60)
print("自定义指标统计汇总")
print("=" * 60)
for name, stats in custom_stats.get_stats().items():
print(f"【{name}】")
print(f" 采集次数: {stats['count']}")
print(f" 平均值: {stats['avg']:.2f}ms")
print(f" 最大值: {stats['max']:.2f}ms")
print(f" 最小值: {stats['min']:.2f}ms")
print(f" 总耗时: {stats['total_sum']:.2f}ms")
print()
3.3 扩展Web界面展示自定义指标
采集到了数据，下一步就是让这些数据在Locust的Web界面上显示出来。Locust的Web界面是Flask应用，它支持自定义页面路由。我们可以加一个新的路由，返回一个JSON接口，然后在界面上显示。
技术栈：Python
# 技术栈：Python
# 文件：web_extension.py
# 扩展Locust Web界面，展示自定义指标
from locust import events
from flask import jsonify, render_template_string
from custom_stats_collector import custom_stats
import json
# 自定义页面的HTML模板
CUSTOM_STATS_TEMPLATE = """
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>自定义指标监控 - Locust</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }
.container { max-width: 1200px; margin: 0 auto; }
h1 { color: #333; border-bottom: 2px solid #4CAF50; padding-bottom: 10px; }
.stats-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 16px; }
.stat-card { background: white; border-radius: 8px; padding: 20px; box-shadow: 0 2px 8px rgba(0,0,0,0.1); }
.stat-name { font-size: 16px; font-weight: bold; color: #4CAF50; margin-bottom: 12px; }
.stat-row { display: flex; justify-content: space-between; margin: 8px 0; font-size: 14px; }
.stat-label { color: #666; }
.stat-value { font-weight: bold; color: #333; }
.refresh-btn { margin: 20px 0; padding: 10px 24px; background: #4CAF50; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 14px; }
.refresh-btn:hover { background: #45a049; }
.refresh-info { font-size: 12px; color: #999; margin-left: 12px; }
</style>
</head>
<body>
<div class="container">
<h1>📊 自定义指标监控面板</h1>
<div>
<button class="refresh-btn" onclick="refreshStats()">🔄 刷新数据</button>
<span class="refresh-info">数据每5秒自动刷新一次</span>
</div>
<div class="stats-grid" id="stats-container">
<!-- JavaScript动态填充内容 -->
</div>
</div>
<script>
// 页面加载时自动刷新
document.addEventListener('DOMContentLoaded', function() {
refreshStats();
// 每5秒自动刷新一次
setInterval(refreshStats, 5000);
});
// 从后端API获取统计数据并渲染
async function refreshStats() {
try {
const response = await fetch('/custom-stats');
const data = await response.json();
renderStats(data);
} catch (error) {
document.getElementById('stats-container').innerHTML = '<p style="color:red;">数据获取失败: ' + error.message + '</p>';
}
}
// 渲染统计卡片
function renderStats(stats) {
const container = document.getElementById('stats-container');
if (Object.keys(stats).length === 0) {
container.innerHTML = '<p style="grid-column: 1/-1; text-align:center; color:#999; padding:40px;">暂无自定义指标数据</p>';
return;
}
let html = '';
for (const [name, stat] of Object.entries(stats)) {
html += `
<div class="stat-card">
<div class="stat-name">${name}</div>
<div class="stat-row"><span class="stat-label">采集次数</span><span class="stat-value">${stat.count}</span></div>
<div class="stat-row"><span class="stat-label">平均值</span><span class="stat-value">${stat.avg.toFixed(2)} ms</span></div>
<div class="stat-row"><span class="stat-label">最大值</span><span class="stat-value">${stat.max.toFixed(2)} ms</span></div>
<div class="stat-row"><span class="stat-label">最小值</span><span class="stat-value">${stat.min.toFixed(2)} ms</span></div>
<div class="stat-row"><span class="stat-label">总耗时</span><span class="stat-value">${stat.total_sum.toFixed(2)} ms</span></div>
</div>
`;
}
container.innerHTML = html;
}
</script>
</body>
</html>
"""
@events.init.add_listener
def add_custom_web_routes(environment, **kwargs):
"""
Locust启动时注册自定义Web路由
这个函数在Locust的Web服务器初始化时被调用
"""
# 获取Flask应用对象
app = environment.runner.web_ui if environment.runner else environment._web_ui
if app is None:
return
# 注册自定义统计数据的API接口
@app.route('/custom-stats')
def custom_stats_api():
"""返回所有自定义指标的JSON数据"""
return jsonify(custom_stats.get_stats())
# 注册自定义统计面板页面
@app.route('/custom-stats-panel')
def custom_stats_panel():
"""渲染自定义指标展示页面"""
return render_template_string(CUSTOM_STATS_TEMPLATE)
3.4 整合所有模块运行
把上面的几个文件整合起来，用一条命令就能启动完整的测试：
技术栈：Python
# 技术栈：Bash命令行
# 启动Locust测试，带上自定义指标采集和Web界面扩展
# 基本启动命令
# --headless 表示无头模式（不弹浏览器），如果需要浏览器交互可以去掉
# --users 指定并发用户数
# --spawn-rate 每秒启动的用户数
# --run-time 测试持续时间
# --host 被测服务的地址
locust -f locustfile.py \
--headless \
--users 50 \
--spawn-rate 10 \
--run-time 5m \
--host http://your-server.com
# 如果想看到Web界面，去掉 --headless 参数，然后浏览器打开 http://localhost:8089
# 自定义面板地址：http://localhost:8089/custom-stats-panel
如果不想用命令行参数，也可以在代码里配置，这样更方便团队协作：
技术栈：Python
# 技术栈：Python
# 文件：settings.py
# Locust全局配置
# 测试目标主机
HOST = "http://your-server.com"
# 并发用户数
NUM_USERS = 50
# 每秒启动的用户数（控制压力增长的速度）
SPAWN_RATE = 10
# 测试持续时间（5分钟）
RUN_TIME = "5m"
# 是否显示HTML格式的统计报告
HTML_STATS = True
技术栈：Python
# 技术栈：Python
# 文件：main.py
# 程序入口：整合所有模块
import locust
from locust import events
from settings import HOST, NUM_USERS, SPAWN_RATE, RUN_TIME
from custom_stats_collector import custom_stats
from web_extension import add_custom_web_routes
# 注册自定义Web路由
events.init.add_listener(add_custom_web_routes)
# 注册测试结束时的数据清理
@events.test_stop.add_listener
def cleanup(environment, **kwargs):
"""测试结束后清理资源"""
# 将统计数据导出到文件，方便后续分析
stats = custom_stats.get_stats()
with open("custom_stats_export.json", "w", encoding="utf-8") as f:
json.dump(stats, f, ensure_ascii=False, indent=2)
print("自定义指标已导出到 custom_stats_export.json")
# 启动Locust
# 使用环境变量传入配置，便于CI/CD集成
if __name__ == "__main__":
import os
os.environ["LOCUST_HOST"] = HOST
os.environ["LOCUST_USERS"] = str(NUM_USERS)
os.environ["LOCUST_SPAWN_RATE"] = str(SPAWN_RATE)
os.environ["LOCUST_RUN_TIME"] = RUN_TIME
locust.main()
四、进阶：把自定义指标接入Prometheus
上面我们做的指标展示只是一个简单的HTML页面，如果你需要和现有的监控体系（比如Grafana）对接，那就要把指标数据以Prometheus格式暴露出来。
Locust其实已经内置了对Prometheus的支持，我们只需要让自定义指标也走Prometheus的格式就行。
技术栈：Python
# 技术栈：Python
# 文件：prometheus_export.py
# 将自定义指标以Prometheus格式暴露
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from locust import events
# 创建Prometheus指标对象
# Counter: 只增不减的计数器（比如请求总数）
custom_request_total = Counter(
"locust_custom_request_total",
"自定义指标累计值",
["name"] # 标签维度
)
# Histogram: 直方图，用于记录耗时分发
custom_response_time = Histogram(
"locust_custom_response_time_seconds",
"自定义指标响应时间分布（秒）",
["name"],
buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0)
)
# Gauge: 可以是任意值的指标（比如当前在线用户数）
custom_gauge_metric = Gauge(
"locust_custom_gauge",
"自定义Gauge指标",
["name"]
)
@events.init.add_listener
def start_prometheus_server(environment, **kwargs):
"""启动Prometheus指标暴露服务"""
# 在9090端口启动Prometheus的metrics端点
start_http_server(9090)
print("Prometheus指标服务已启动，访问 http://localhost:9090/metrics")
@events.custom.add_listener
def push_to_prometheus(environment, **kwargs):
"""
将自定义事件数据推送到Prometheus
注意：Prometheus的Counter和Histogram的标签（labels）数量有限
"""
name = kwargs.get("name", "unknown")
value = kwargs.get("value", 0)
# 更新Counter
custom_request_total.labels(name=name).inc()
# 更新Histogram（传入耗时，单位是秒）
if isinstance(value, (int, float)):
custom_response_time.labels(name=name).observe(float(value) / 1000.0)
对应的Grafana中可以用PromQL查询：
技术栈：Text（PromQL查询示例）
# 技术栈：PromQL查询语言
# 查询自定义指标的平均响应时间（按名称分组）
rate(locust_custom_response_time_seconds_sum{name="订单创建耗时"}[5m]) / rate(locust_custom_response_time_seconds_count{name="订单创建耗时"}[5m])
# 查询自定义指标的p99响应时间
histogram_quantile(0.99, sum(rate(locust_custom_response_time_seconds_bucket[5m])) by (le, name))
# 查询自定义指标的总请求次数
sum(increase(locust_custom_request_total[1h])) by (name)
Grafana的dashboard配置示例：
技术栈：JSON
{
"title": "Locust自定义指标监控",
"panels": [
{
"title": "各指标请求次数（每小时）",
"type": "bar",
"datasource": "Prometheus",
"targets": [
{
"expr": "sum(increase(locust_custom_request_total[1h])) by (name)",
"legendFormat": "{{name}}"
}
],
"fieldConfig": {
"defaults": {
"unit": "short",
"color": { "mode": "palette-classic" }
}
}
},
{
"title": "响应时间P99趋势",
"type": "timeSeries",
"targets": [
{
"expr": "histogram_quantile(0.99, sum(rate(locust_custom_response_time_seconds_bucket[5m])) by (le, name))",
"legendFormat": "{{name}} - P99"
},
{
"expr": "histogram_quantile(0.50, sum(rate(locust_custom_response_time_seconds_bucket[5m])) by (le, name))",
"legendFormat": "{{name}} - P50"
}
],
"fieldConfig": {
"defaults": {
"unit": "s",
"custom": {
"drawStyle": "line",
"lineInterpolation": "smooth",
"fillOpacity": 10,
"spanNulls": true
}
}
}
}
]
}
五、应用场景分析
这套自定义指标采集方案，在以下几种场景下特别有用：
业务链路追踪场景：电商下单、支付、物流这几种核心流程，每一步都涉及多个微服务调用。光看HTTP层面的状态码和响应时间根本不够，你需要知道"库存校验花了多少毫秒"、"订单服务调用了多少次物流服务"这种细粒度的信息。
性能瓶颈定位场景：测试跑完之后，整体P95响应时间是2秒，但到底慢在哪里？是数据库查询慢、是第三方接口慢、还是你自己的业务逻辑有性能问题？自定义指标能把每个环节拆开来看。
容量规划场景：在扩容决策的时候，光看请求量是不够的。你需要知道不同业务模块的负载差异，比如"商品搜索接口QPS是1000，订单创建接口QPS只有100，但后者是前者的50倍耗时"，这种信息才能指导你怎么分配资源。
回归测试对比场景：每次上线新版本后，除了看有没有bug，还要看性能有没有退化。有了自定义指标，你可以精确对比"上线前商品详情接口平均耗时120ms，上线后变成180ms"，及时发现性能问题。
六、技术优缺点对比
说完了好处，也得说说这套方案有什么坑。
优点方面：
第一，非侵入式。不需要改Locust源码，不需要fork项目，只需要加几个Python文件就能实现。升级Locust版本的时候也不用担心兼容性问题。
第二，灵活度很高。采集什么指标、怎么采集，完全由你决定。可以按业务需要动态调整，不需要改框架。
第三，和现有工具链兼容。Prometheus和Grafana都是行业标准，接入之后可以和现有的监控告警体系打通。
缺点方面：
第一，内存开销。自定义采集器在内存里维护统计数据，如果指标种类太多或者采集频率太高，内存占用会增长。好在我们在采集器里加了1000条时间戳的上限保护，一般不会出问题。
第二，没有内置持久化。我们方案里的数据都是存在内存里的，测试结束后如果不主动导出就会丢失。如果需要长期存储，得配合Prometheus或者自己写数据库持久化逻辑。
第三，Web界面扩展有限。Locust原生的Web界面是Flask应用，我们加的路由能工作，但样式和原生界面不太统一，看起来会有点割裂。
七、注意事项
线程安全问题：Locust是多线程的，多个线程可能同时发事件。采集器里用threading.Lock做了保护，但如果你自己写采集逻辑，一定要记得加锁。不加锁的话，统计数据可能会出现脏读、数据不一致的问题。
指标名称命名规范：自定义指标的名称建议用中文或者带下划线的英文，避免使用特殊字符。名称会作为Prometheus的label值，特殊字符可能导致Prometheus解析失败。推荐格式：模块名_操作类型_度量单位，比如订单_创建耗时_ms。
性能开销：每次发事件都会触发一次事件总线的分发，虽然开销很小（微秒级别），但在超高QPS场景下（每秒几万请求以上）累积起来也是可观的。如果性能敏感，可以考虑批量上报，比如攒够100条再统一推送。
清理机制：Locust的事件监听器在测试过程中是持续有效的，但如果你多次启动测试（比如在同一次locust运行中重复start/stop），自定义采集器的数据可能会累积。建议在测试开始时调用reset()方法清空旧数据。
环境变量注入：在CI/CD环境中，通过环境变量传入配置比硬编码更灵活。Locust支持用LOCUST_HOST、LOCUST_USERS等环境变量覆盖命令行参数，自定义采集器也可以读取环境变量来控制行为。
八、文章总结
回到开头的问题：Locust的Web界面信息不全怎么办？
答案是，别指望Locust帮你展示所有东西，把自定义指标的采集和展示交给自己写代码去解决。核心思路就是三步：
第一步，用events.custom.fire()发事件，把你在测试脚本里计算出来的业务指标往事件总线上扔。
第二步，写一个采集器类，监听事件总线上的数据，做统计和汇总。
第三步，通过自定义Flask路由把数据展示在Web界面上，或者推送到Prometheus让Grafana来展示。
整个方案的关键在于利用Locust的事件总线机制。这个机制设计得很干净，不需要深入改源码就能实现扩展，而且和Locust的版本更新解耦。
如果你的测试场景很简单，只需要跑个接口看状态码和响应时间，那Locust原生功能就够用了。但如果你需要追踪业务级别的指标——耗时分布、成功率、调用次数、业务流水号——那就必须动手加自定义采集。
最后给个建议：如果你的项目是正经的产品级压测，建议从一开始就把自定义指标采集的框架搭好，不要等测试跑了几次觉得"好像还差点什么"才想起加。指标采集越早集成，后续的分析价值越大。
评论
围绕“Locust Web界面监控信息不全？扩展自定义指标的采集与展示方案”参与讨论