如何基于Locust自定义负载形状实现预定义分布的每秒请求发送

核心思路转换 #

Locust本身是用户驱动的压测模型，要实现固定RPS的负载，需要先建立「用户数-单用户RPS-总RPS」的对应关系：

• 先测试单用户在无额外逻辑、只发目标请求的场景下的稳定RPS值
• 总目标RPS = 运行用户数 * 单用户稳定RPS，由此可以反推出需要的用户数

前置准备 #

• 编写最小化测试脚本，固定1个用户运行3-5分钟，统计平均RPS，记为PER_USER_RPS，建议多次测试取平均值，保证数值稳定
• 按照你的需求生成RPS池列表，列表中每个元素对应1秒的目标RPS，数值出现次数和Apache日志中的统计结果完全一致

完整实现代码 #

from locust import LoadTestShape, HttpUser, task, constant_pacing
import random

# -------------------------- 配置项 --------------------------
# 替换为你实际测试得到的单用户稳定RPS
PER_USER_RPS = 2
# 替换为你从Apache日志统计生成的RPS池列表
RPS_POOL = [1,2,2,3,5,5,5,18,18,36]
# 固定随机种子，保证每次测试的RPS序列完全一致，实现可复现
RANDOM_SEED = 42
# 测试总时长，单位秒，对应5分钟
TEST_DURATION = 300
# -----------------------------------------------------------

random.seed(RANDOM_SEED)

class TestUser(HttpUser):
    # 固定请求间隔，保证单用户RPS稳定为PER_USER_RPS
    wait_time = constant_pacing(1 / PER_USER_RPS)
    
    @task
    def send_target_request(self):
        # 替换为你实际需要压测的接口地址
        self.client.get("/your/target/path")

class RpsDistributeLoadShape(LoadTestShape):
    def tick(self):
        run_time = self.get_run_time()
        # 超过测试时长则结束压测
        if run_time >= TEST_DURATION:
            return None
        # 随机选取当前秒的目标RPS
        current_target_rps = random.choice(RPS_POOL)
        # 计算需要的用户数，最少保留1个用户
        target_user_count = max(1, round(current_target_rps / PER_USER_RPS))
        # 孵化速率设置为足够大，保证1秒内可以完成用户数的调整
        spawn_rate = target_user_count * 2
        return (target_user_count, spawn_rate)

效果说明 #

• 该方案实现的实际RPS会和目标值有±10%以内的波动，完全满足你对比API服务器配置变更效果的需求
• 固定随机种子后，每次测试的RPS序列完全一致，多次测试的负载可复现

优化建议 #

• 如果发现实际RPS和目标值偏差较大，可以重新校准PER_USER_RPS的取值
• 如果目标RPS波动幅度很大，可以进一步调大spawn_rate的倍率，比如设置为100，保证用户数可以快速跟上目标调整

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