如何在locustfile.py中将Locust的/stats/requests数据发往第三方服务绘图

实现方案 #

你之前的代码之所以会阻塞进程，是因为死循环直接运行在了Locust的主线程中，正确做法是把上报逻辑放到独立的后台守护线程中，通过Locust自带的事件钩子让线程在测试启动时自动运行。除此之外你不需要主动请求/stats/requests接口，直接从Locust的运行时环境就能拿到全量统计数据，和接口返回结构完全一致，效率更高。

完整实现代码如下：

from locust import HttpUser, task, events
import requests
import threading
import time

# 自定义配置：上报间隔（单位秒）、上报目标地址
REPORT_INTERVAL = 5
REPORT_TARGET = "http://another-server-for-plotting"

class User(HttpUser):
    @task
    def scenario_01(self):
        self.client.get('/some_endpoint')

def report_worker(environment):
    """后台上报线程逻辑"""
    # 压测停止后自动结束循环
    while environment.runner.state != "stopped":
        # 直接从运行环境获取统计数据，和/stats/requests返回结构完全一致
        stats = environment.runner.stats.serialize()
        # 可按需额外补充字段，比如当前并发用户数
        payload = {
            **stats,
            "current_users": environment.runner.user_count
        }
        try:
            requests.post(REPORT_TARGET, json=payload, timeout=2)
        except Exception:
            # 上报异常不影响压测主流程，可按需补充日志打印逻辑
            pass
        time.sleep(REPORT_INTERVAL)

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """测试启动时触发，初始化上报线程"""
    # 非Worker节点才运行上报逻辑，分布式部署时避免多worker节点重复上报，单机运行不受影响
    if environment.runner.__class__.__name__ != "WorkerRunner":
        # 设为守护线程，Locust进程退出时线程会自动销毁
        report_thread = threading.Thread(target=report_worker, args=(environment,), daemon=True)
        report_thread.start()

补充说明 #

• 上报逻辑加了异常捕获，上报失败不会打断压测流程，可根据需要自行添加日志记录异常
• 如果不需要自定义字段，直接上报environment.runner.stats.serialize()的返回值即可，和/stats/requests接口返回的数据完全一致
• 代码兼容单机、分布式两种部署模式，无需额外修改

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