Locust固定RPS负载测试：如何获取当前RPS与用户数？

你可以通过Locust的LoadTestShape类内置的self.runner属性获取当前运行时的用户数和RPS数据，同时需要在自定义负载类中添加状态变量来记录RPS达标时的用户数，实现你的测试场景。

完整修改后的代码 #

from locust import TaskSet, HttpUser, LoadTestShape

class Scan(TaskSet):
    @task
    def perf_scan(self):
        self.client.get("/api")

class UrlUser(HttpUser):
    tasks = [Scan]

class CustomLoadShapeScenario2(LoadTestShape):
    def __init__(self):
        super().__init__()
        # 初始化变量，记录RPS达标时的用户数
        self._target_users = None

    def tick(self):
        total_time = 3600  # 总时长60分钟
        run_time = self.get_run_time()
        # 获取当前用户数和RPS
        current_users = self.runner.user_count
        current_rps = self.runner.stats.total.current_rps

        # 前10分钟（600秒）内，线性增加用户直到RPS≥70
        if run_time < 600:
            if current_rps < 70:
                # 每10秒增加1个用户，线性增长
                users = int(run_time // 10)
                spawn_rate = 1
            else:
                # RPS达标，记录当前用户数，进入稳态
                self._target_users = current_users
                users = self._target_users
                spawn_rate = 0
        elif run_time < total_time:
            # 总时长未到，维持达标时的用户数
            if self._target_users is not None:
                users = self._target_users
                spawn_rate = 0
            else:
                # 极端情况：10分钟内RPS仍未达标，继续按原逻辑增加用户
                users = int(run_time // 10)
                spawn_rate = 1
        else:
            # 测试结束，停止生成用户
            return None

        return (users, spawn_rate)

关键说明 #

• 获取当前用户数：通过self.runner.user_count直接获取Locust当前运行的用户总数
• 获取当前RPS：通过self.runner.stats.total.current_rps获取所有请求的实时RPS数据
• 状态记录：用self._target_users存储RPS首次达标时的用户数，确保后续稳态阶段维持该用户规模
• 逻辑优化：区分前10分钟的增长阶段和后续稳态阶段，同时处理10分钟内RPS未达标的极端情况

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