如何用Locust负载测试实现固定RPS调用指定HTTP GET API

你需要调用/hello接口并保持50请求/秒的恒定频率，但默认的Locust任务写法会因为API响应时间波动导致RPS不稳定，下面提供几种可行的实现方式：

方法1：用constant_pacing实现近似恒定RPS #

这种方式适合响应时间稳定的场景，它控制两次任务执行的间隔，让请求触发频率尽量接近目标值。如果API响应时间短于设定间隔，会自动等待到指定时间再发起下一次请求；但如果响应时间超过间隔，RPS会暂时低于目标。

代码示例：

from locust import HttpUser, task, constant_pacing

class HelloWorldUser(HttpUser):
    # 设定两次任务启动的间隔为0.02秒（1/50），目标每秒50次请求
    wait_time = constant_pacing(1/50)

    @task
    def hello_world(self):
        self.client.get("/hello")

方法2：自定义等待逻辑实现严格恒定RPS #

如果需要严格保证RPS不受响应时间影响，可以手动计算每次请求的等待时间，确保请求发送频率精准匹配目标。

代码示例（单用户实现50 RPS，多用户需按比例分配）：

from locust import HttpUser, task
import time

class HelloWorldUser(HttpUser):
    target_rps = 50
    last_request_time = 0

    def on_start(self):
        self.last_request_time = time.time()

    @task
    def hello_world(self):
        current_time = time.time()
        # 计算需要等待的时间，确保每秒发送指定数量的请求
        elapsed = current_time - self.last_request_time
        expected_interval = 1 / self.target_rps
        wait_time = max(0, expected_interval - elapsed)
        
        time.sleep(wait_time)
        
        self.client.get("/hello")
        # 更新最后一次请求的发送时间
        self.last_request_time = time.time()

如果需要用多个用户分担请求，比如5个用户，就把每个用户的target_rps设为10，总RPS就能保持50。

方法3：借助第三方插件快速实现 #

如果允许使用第三方工具，locust-plugins中的ConstantRPSUser可以直接全局控制RPS，无需手动计算：

代码示例：

from locust import task
from locust_plugins.users import ConstantRPSUser

class HelloWorldUser(ConstantRPSUser):
    # 直接设置全局目标RPS
    constant_rps = 50

    @task
    def hello_world(self):
        self.client.get("/hello")

使用前需要先安装插件：

pip install locust-plugins

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