如何在Locust中基于LoadTestShape类设置每分钟事务数

要实现固定每分钟事务数（TPM）的负载测试，仅靠固定user_count无法达成目标——事务速率（RPS/TPM）会受CPU、网络、任务执行耗时等因素波动。以下是两种可靠的实现方式：

方法一：使用内置的FixedRateShape（Locust 2.x+推荐） #

Locust的FixedRateShape类会自动动态调整用户数量，确保系统维持设定的每秒事务数（RPS），从而稳定实现目标TPM。

from locust import LoadTestShape

class FixedTPMLoadShape(LoadTestShape):
    # 将每分钟1000事务换算为每秒事务数：1000/60 ≈16.67
    rate = 1000 / 60
    time_limit = 3600  # 测试总时长（秒），按需修改
    spawn_rate = 10  # 每秒新增/减少的用户数，按需调整

    def tick(self):
        run_time = self.get_run_time()
        if run_time < self.time_limit:
            # 返回（目标用户数，生成速率），FixedRateShape会自动适配用户数维持设定rate
            return (self.get_current_user_count(), self.spawn_rate)
        return None

原理说明 #

FixedRateShape会持续监控实际事务速率，当速率低于设定值时自动增加用户，超过设定值时减少用户，以此稳定维持目标TPM。

方法二：自定义LoadTestShape+任务速率控制 #

如果需要更精细的控制，可以结合自定义Shape类和任务内的固定间隔设置：

1. 自定义Shape类动态调整用户数 #

from locust import LoadTestShape

class TargetTPMShape(LoadTestShape):
    target_tpm = 1000
    target_rps = target_tpm / 60
    # 预先压测得到的单任务平均执行时长（秒，含网络请求+处理时间）
    avg_task_duration = 1.2
    spawn_rate = 5

    def tick(self):
        run_time = self.get_run_time()
        # 计算维持目标RPS所需的用户数：RPS × 单任务平均时长
        required_users = int(self.target_rps * self.avg_task_duration)
        current_users = self.get_current_user_count()

        if run_time < 3600:
            # 逐步将用户数调整到目标值
            if current_users < required_users:
                return (required_users, self.spawn_rate)
            else:
                return (current_users, 0)
        return None

2. 在用户类中设置固定任务间隔 #

通过constant_pacing控制每个用户的任务执行间隔，确保单个用户的事务速率稳定：

from locust import HttpUser, task, constant_pacing

class MyTestUser(HttpUser):
    # 每个任务间隔60/1000秒，确保单用户每分钟执行1000次任务（需保证任务耗时不超过该间隔）
    wait_time = constant_pacing(60 / 1000)

    @task
    def core_transaction(self):
        # 你的核心事务逻辑，比如接口请求
        self.client.get("/api/your-endpoint")

注意事项 #

• avg_task_duration需要通过预压测获取真实值，否则计算出的用户数会偏离目标。
• 如果任务耗时超过constant_pacing设置的间隔，Locust会忽略等待时间立即执行下一个任务，此时需要通过Shape类增加用户数来维持目标TPM。

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