API负载测试能否控制请求数？Locust任务请求数限制求助

原代码的问题 #

你当前的代码里，counter是UserJourney的类属性，所有用户实例会共享这个值——只要有多个用户同时运行，计数器会被快速累加，远没到1000次请求就会停止；而且不管是否发送了请求，每次任务执行都会加1，逻辑也不准确。

两种可行的解决方案 #

方案1：控制全局总请求数 #

如果要让整个测试总共发送指定数量的请求（比如1000次），可以用Locust的事件监听器配合进程安全的计数器实现：

from locust import SequentialTaskSet, HttpUser, task, events
from multiprocessing import Value

# 全局请求计数器，用Value保证多进程/线程安全
total_requests = Value('i', 0)
# 设置总请求限制
REQUEST_LIMIT = 1000

def update_counter(request_type, name, response_time, response_length, exception, context):
    # 每次请求完成（成功/失败）都累加计数器
    with total_requests.get_lock():
        total_requests.value += 1
        # 达到限制后停止整个测试
        if total_requests.value >= REQUEST_LIMIT:
            runner = events.runner
            if runner:
                runner.quit()

# 注册请求事件监听器
events.request_success.add_listener(update_counter)
events.request_failure.add_listener(update_counter)

class UserJourney(SequentialTaskSet):
    @task
    def demo_get_test(self):
        self.client.get("/api/users?page=2", name='Users List')

class DemoHttpUser(HttpUser):
    host = "http://reqres.in"
    tasks = [UserJourney]

这个方案的逻辑是：每次请求完成后自动累加全局计数器，达到设定的总请求数时，调用runner.quit()停止所有Locust进程，整个测试结束。

方案2：控制每个用户的请求数 #

如果要让每个用户只发送指定数量的请求（比如每个用户发10次），可以把计数器改成用户实例的属性，达到限制后中断该用户的任务：

from locust import SequentialTaskSet, HttpUser, task

class UserJourney(SequentialTaskSet):
    def on_start(self):
        # 每个用户实例单独维护自己的计数器
        self.user_counter = 0
        # 单个用户的请求限制
        self.user_request_limit = 10

    @task
    def demo_get_test(self):
        if self.user_counter < self.user_request_limit:
            self.client.get("/api/users?page=2", name='Users List')
            self.user_counter += 1
        else:
            # 该用户完成所有请求，停止任务集
            self.interrupt()

class DemoHttpUser(HttpUser):
    host = "http://reqres.in"
    tasks = [UserJourney]

这个方案里，每个用户启动时初始化自己的计数器，每次成功发送请求后累加，达到限制后调用self.interrupt()让该用户停止任务，不再发送请求。

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