如何在Locust.io的Python任务中通过LoadTestShape指定自定义参数？

方法一：通过LoadTestShapes共享阶段参数 #

虽然LoadTestShapes的tick()只能返回用户数、生成速率和用户类，但可以在Shape类里维护自定义阶段配置，再把当前阶段的参数存入Locust的环境变量中，让HttpUser读取使用。

示例代码：

from locust import HttpUser, task, LoadTestShape

class CustomLoadShape(LoadTestShape):
    # 定义各阶段的配置：(持续时间秒, 用户数, 生成速率, 自定义参数)
    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 1, "custom_param": "config_A"},
        {"duration": 120, "users": 20, "spawn_rate": 2, "custom_param": "config_B"},
        {"duration": 180, "users": 30, "spawn_rate": 3, "custom_param": "config_C"},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stage["duration"]:
                # 将当前阶段的自定义参数存入环境的user_data
                self.environment.user_data["current_param"] = stage["custom_param"]
                return (stage["users"], stage["spawn_rate"])
        return None

class MyUser(HttpUser):
    @task
    def test_task(self):
        # 从环境中获取当前阶段的自定义参数
        current_param = self.environment.user_data.get("current_param", "default_config")
        # 用参数发起请求，比如作为查询参数或请求体
        self.client.get(f"/api/test?config={current_param}")

方法二：在HttpUser中直接通过运行时间判断阶段 #

你提到无法在任务内访问get_run_time()，其实在@task装饰的方法里，self就是HttpUser的实例，直接调用self.get_run_time()就能获取运行时长，据此判断当前阶段并切换参数。

示例代码：

from locust import HttpUser, task

class MyUser(HttpUser):
    @task
    def test_task(self):
        run_time = self.get_run_time()
        # 根据运行时间划分阶段，设置对应参数
        if run_time < 60:
            custom_param = "config_A"
        elif run_time < 120:
            custom_param = "config_B"
        elif run_time < 180:
            custom_param = "config_C"
        else:
            custom_param = "default_config"
        
        # 使用参数执行请求
        self.client.post("/api/action", json={"config": custom_param})

方法三：用全局变量+定时器同步阶段（适合简单场景） #

如果测试逻辑不复杂，可以定义全局变量存储当前参数，再用一个后台线程根据运行时间更新参数。注意线程安全，建议结合Locust的事件钩子初始化。

示例代码：

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

current_custom_param = "config_A"

def update_param():
    time.sleep(60)
    global current_custom_param
    current_custom_param = "config_B"
    time.sleep(60)
    current_custom_param = "config_C"

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    # 启动线程更新参数
    threading.Thread(target=update_param, daemon=True).start()

class MyUser(HttpUser):
    @task
    def test_task(self):
        global current_custom_param
        self.client.get(f"/api/test?config={current_custom_param}")

内容的提问来源于stack exchange，提问作者Luca P.