关于Locust实时图表导出及数据获取存储的技术问询

一、将实时图表导入自研UI #

Locust自带REST API，可直接获取实时测试指标，你完全可以在自研UI中调用这些接口，自行渲染图表，无需依赖Locust原生界面：

• 核心API端点：
  • GET /stats/requests：返回所有请求的实时统计数据（含响应时间、成功率、每秒请求数等）
  • GET /stats/failures：返回失败请求的实时统计
• 实现思路：在自研UI中通过定时轮询（比如每1-5秒）调用上述接口，拿到JSON格式数据后，用ECharts、Chart.js等前端图表库绘制实时折线图、柱状图等
• 进阶方案：如果需要更低延迟的实时推送，可基于Locust的事件系统自定义扩展——监听request_success、request_failure等事件，通过WebSocket将数据主动推送到自研UI

二、实时数据导出至本地设备 #

如果暂时无法集成图表，以下几种方式可实现实时数据本地导出：

1. 自定义事件监听插件 #

编写简单的Locust扩展，监听请求事件并实时写入本地文件：

from locust import events
import csv
from datetime import datetime

class RealtimeDataExporter:
    def __init__(self):
        # 初始化CSV文件
        self.csv_file = open("locust_realtime_data.csv", "w", newline="", encoding="utf-8")
        self.writer = csv.writer(self.csv_file)
        # 写入表头
        self.writer.writerow(["时间戳", "请求类型", "接口名称", "响应时间(ms)", "是否成功"])
        # 绑定事件监听
        events.request_success.add_listener(self.on_request_success)
        events.request_failure.add_listener(self.on_request_failure)
        events.quitting.add_listener(self.on_test_end)

    def on_request_success(self, request_type, name, response_time, response_length):
        self.writer.writerow([datetime.now().isoformat(), request_type, name, round(response_time, 2), "是"])
        self.csv_file.flush()  # 实时写入磁盘

    def on_request_failure(self, request_type, name, response_time, exception):
        self.writer.writerow([datetime.now().isoformat(), request_type, name, round(response_time, 2), "否"])
        self.csv_file.flush()

    def on_test_end(self, environment, **kwargs):
        self.csv_file.close()

# 在Locustfile中初始化
data_exporter = RealtimeDataExporter()

2. 使用Locust内置CSV导出 #

启动Locust时添加--csv=测试结果参数，测试过程中会自动生成两个实时更新的CSV文件：

• 测试结果_requests.csv：记录所有请求的统计数据
• 测试结果_failures.csv：记录所有失败请求的详情
  数据会随测试进行持续写入文件，无需额外编码

3. 日志解析方案 #

配置Locust的日志格式，将请求关键信息输出到日志文件，后续可通过脚本解析日志提取数据，但这种方式不如结构化文件直观，适合临时需求

内容的提问来源于stack exchange，提问作者Siddesh Ubale