Locust v2.20.1中self.stop()失效,如何实现单用户单次任务执行?
Locust实现n个用户各执行一次任务的解决方案
针对Locust v2.20.1中生成指定数量用户、每个用户仅执行一次任务的需求,可通过以下方式实现,同时避免你遇到的单个用户停止报错问题:
核心解决方案:安全停止单个用户
不要使用self.stop()终止用户,改用Locust提供的stop_user()方法,该方法能安全终止单个用户进程,不会出现参数错误问题。
方法1:在用户启动时执行一次性任务
将所有任务放在用户的on_start方法中,执行完成后直接停止当前用户:
from locust import HttpUser class OneTimeUser(HttpUser): def on_start(self): # 按需求执行所有单次任务 self.client.get("/api/endpoint1") self.client.post("/api/endpoint2", json={"data": "test"}) # 安全停止当前用户 self.environment.runner.stop_user(self) # 无需定义task,任务全部在on_start中执行 tasks = []
方法2:用任务序列执行后停止用户
如果需要按顺序执行多个任务,可使用SequentialTaskSet,在最后一个任务中停止当前用户:
from locust import HttpUser, SequentialTaskSet, task class SingleRunTasks(SequentialTaskSet): @task def first_task(self): self.client.get("/step1") @task def second_task(self): self.client.post("/step2") # 任务完成后停止当前用户 self.environment.runner.stop_user(self.user) class OneTimeUser(HttpUser): tasks = [SingleRunTasks] # 取消任务间等待时间,直接执行 wait_time = None
运行测试
启动测试时指定用户数量即可,所有用户完成任务后会自动终止测试:
locust -f your_script.py --users 10 --headless
为什么之前的方法出错?
你调用的self.stop()是TaskSet类的方法,仅用于停止当前任务集合,而非终止整个用户进程。直接用它停止用户会导致Locust内部参数异常,出现你遇到的didn't have proper args错误。而stop_user()是Locust runner提供的官方方法,专门用于安全终止单个用户实例。
内容的提问来源于stack exchange,提问作者Alexander Telecan
相关产品推荐
相关产品推荐

