Locust技术咨询：单用户请求数配置与多用户带唯一Token并发测试

1. Locust 是否支持配置每个用户发送n次请求？ #

当然支持，你可以通过多种方式精准控制每个用户的请求执行次数：

• 固定次数循环：在任务方法中添加循环，指定每个用户执行请求的次数
• 自定义执行逻辑：结合用户生命周期方法和任务装饰器，灵活控制请求次数
• 参数化配置：通过外部文件或环境变量动态设置每个用户的请求次数

示例代码（每个用户发送5次请求）：

from locust import HttpUser, task, between

class MultiRequestUser(HttpUser):
    wait_time = between(1, 2)  # 可自定义请求间隔

    @task(1)
    def repeat_requests(self):
        token = "your_token_here"  # 实际场景可从Token池获取
        target_id = 123
        # 每个用户固定发送5次请求
        for _ in range(5):
            headers = {'Authorization': f'Bearer {token}'}
            self.client.put(f"/api/{target_id}/like", headers=headers)

2. 实现n个用户用唯一Token并发发起单次请求 #

要实现n个用户、每个用唯一Token、仅发一次请求、同时发起的需求，核心是用队列管理Token池，结合Locust的用户生命周期和启动速率配置，具体步骤如下：

步骤1：准备Token池与队列管理 #

用队列存储所有唯一Token，确保每个用户启动时获取不重复的Token：

from locust import HttpUser, task, events
from queue import Queue

# 初始化Token队列，实际场景可从文件/数据库读取
token_queue = Queue()

def setup_token_pool():
    # 替换为你的n个唯一Token，示例从列表导入
    for token in ["token_001", "token_002", "token_003", ...]:
        token_queue.put(token)
    # 若从文件读取可替换为：
    # with open("tokens.txt", "r") as f:
    #     for line in f:
    #         token_queue.put(line.strip())

# 测试启动前初始化Token池
events.init.add_listener(setup_token_pool)

步骤2：编写用户类，单次请求后停止 #

每个用户启动时获取唯一Token，执行一次请求后立即停止，避免重复执行：

class ConcurrentSingleRequestUser(HttpUser):
    wait_time = 0  # 无等待，确保请求立即发送
    target_id = 123  # 替换为实际目标ID，若每个用户对应不同ID可新增队列管理

    def on_start(self):
        # 用户启动时从队列获取唯一Token
        self.token = token_queue.get()

    @task(1)
    def send_single_like_request(self):
        headers = {
            'Authorization': f'Bearer {self.token}'
        }
        # 执行请求
        self.client.put(f"/api/{self.target_id}/like", headers=headers)
        # 请求完成后停止当前用户，不再执行后续任务
        self.stop()

步骤3：配置测试参数实现并发 #

运行测试时，在Locust UI或命令行设置：

• 用户数（Number of users）：设为n（你需要的并发用户数）
• 孵化速率（Spawn rate）：设为n（瞬间生成所有n个用户，确保请求同时发起）

命令行示例：

locust -f your_test_file.py --users n --spawn-rate n --headless -t 1m

关键说明 #

• 队列确保每个用户获取唯一Token，避免重复使用
• wait_time = 0 取消用户请求间隔，保证请求立即发送
• self.stop() 让用户执行完一次请求后停止，符合"仅发送一次请求"的要求
• 孵化速率等于用户数，实现所有用户同时启动，请求同时发起，精准模拟并发压力

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