如何在Locust测试中获取每秒请求数并在用户线程执行前后记录？

核心原理 #

Locust的统计数据（包括请求数、响应时间等）都维护在locust.stats模块中，图表展示的RPS分为两种：累计RPS（总请求数/总测试时间）和实时滑动窗口RPS（最近N秒内的请求数/窗口时长）。我们可以通过直接调用统计模块或结合事件监听来获取这些数据。

方法1：获取累计RPS（用户线程前后） #

直接利用locust.stats提供的全局统计变量，在用户线程的启动/结束钩子中计算累计RPS：

from locust import User, task
from locust.stats import stats

class TestUser(User):
    def on_start(self):
        # 线程启动时计算累计RPS
        total_rps = stats.total_requests / stats.total_time if stats.total_time > 0 else 0
        print(f"线程启动 - 累计RPS: {total_rps:.2f}")
        
        # 也可以获取单个请求类型的累计RPS
        for request_name, entry in stats.request_stats.items():
            entry_rps = entry.num_requests / entry.total_time if entry.total_time > 0 else 0
            print(f"请求[{request_name}]累计RPS: {entry_rps:.2f}")

    def on_stop(self):
        # 线程结束时计算累计RPS
        total_rps = stats.total_requests / stats.total_time if stats.total_time > 0 else 0
        print(f"线程结束 - 累计RPS: {total_rps:.2f}")

    @task
    def sample_task(self):
        # 你的业务请求逻辑
        self.client.get("/api/test")

方法2：获取实时滑动窗口RPS（与图表数据一致） #

Locust图表中的RPS是基于滑动窗口的实时数据，我们可以通过监听请求事件、维护时间戳队列来模拟这个逻辑：

from locust import User, task, events
import time

# 定义滑动窗口时长（与Locust图表默认一致，比如10秒）
SLIDING_WINDOW = 10
# 存储请求时间戳的队列
request_timestamps = []

def track_requests(event, **kwargs):
    # 记录每个请求的时间戳
    request_timestamps.append(time.time())
    # 清理窗口外的旧数据
    cutoff_time = time.time() - SLIDING_WINDOW
    while request_timestamps and request_timestamps[0] < cutoff_time:
        request_timestamps.pop(0)

# 绑定请求成功/失败事件，跟踪所有请求
events.request_success.add_listener(track_requests)
events.request_failure.add_listener(track_requests)

class TestUser(User):
    def on_start(self):
        # 计算窗口内的实时RPS
        realtime_rps = len(request_timestamps) / SLIDING_WINDOW
        print(f"线程启动 - 实时RPS（最近{SLIDING_WINDOW}秒）: {realtime_rps:.2f}")

    def on_stop(self):
        realtime_rps = len(request_timestamps) / SLIDING_WINDOW
        print(f"线程结束 - 实时RPS（最近{SLIDING_WINDOW}秒）: {realtime_rps:.2f}")

    @task
    def sample_task(self):
        self.client.get("/api/test")

分布式测试场景下的处理 #

如果是分布式运行Locust，单个节点的统计数据是局部的，需要从主节点获取汇总后的统计：

from locust import User, task
from locust.stats import stats
from locust.runners import MasterRunner, runners

def get_global_rps():
    if isinstance(runners.locust_runner, MasterRunner):
        # 从主节点获取全局累计统计
        total_reqs = runners.locust_runner.stats.total_requests
        total_time = runners.locust_runner.stats.total_time
    else:
        # 单机模式直接用本地统计
        total_reqs = stats.total_requests
        total_time = stats.total_time
    return total_reqs / total_time if total_time > 0 else 0

class TestUser(User):
    def on_start(self):
        global_rps = get_global_rps()
        print(f"线程启动 - 全局累计RPS: {global_rps:.2f}")

    @task
    def sample_task(self):
        self.client.get("/api/test")

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