如何实现Locust Master为Slave动态分配用户及动态URL压测

嘿，这个需求我熟！要实现用Locust从CSV读取不同URL和对应模拟用户数的负载测试，咱们可以通过动态生成用户类+权重配置的方式搞定，刚好匹配你的需求。下面是具体步骤和完整脚本：

1. 核心思路 #

Locust的用户类支持weight属性，这个属性决定了不同用户类的用户数比例。咱们可以把CSV里的“需模拟的用户数”直接作为对应URL用户类的权重，然后启动时设置总用户数为所有用户数的总和，这样每个URL的模拟用户数就完全符合你的要求啦。

2. 完整实现脚本 #

import csv
from locust import HttpUser, task, between

# 读取CSV文件，加载URL和对应用户数
def load_url_data():
    url_data = []
    with open("Input.csv", "r", encoding="utf-8") as f:
        # 适配制表符分隔的CSV（你的示例是tab分隔），如果是逗号分隔就把delimiter改成","
        reader = csv.DictReader(f, delimiter="\t")
        for row in reader:
            url_data.append({
                "url": row["URLs"].strip(),
                "user_count": int(row["需模拟的用户数"].strip())
            })
    return url_data

url_list = load_url_data()

# 动态生成每个URL对应的用户类
for entry in url_list:
    target_url = entry["url"]
    user_weight = entry["user_count"]
    
    # 生成合法的类名，替换URL里的特殊字符（避免Python语法错误）
    safe_class_name = f"LoadTestUser_{target_url.replace('http://', '').replace('https://', '').replace('/', '_').replace('.', '_').replace(':', '_')}"
    
    # 定义针对当前URL的GET请求任务
    def create_get_task(url):
        def task_method(self):
            # 用name参数聚合相同路径的请求（如果URL有动态参数，避免报告分散）
            self.client.get(url, name=url.split('?')[0])
        return task_method
    
    # 动态创建用户类
    user_class = type(
        safe_class_name,
        (HttpUser,),
        {
            "wait_time": between(1, 3),  # 自定义用户操作间隔，可根据需求调整
            "weight": user_weight,
            "task": create_get_task(target_url)
        }
    )
    
    # 将类注册到全局，让Locust能够识别
    globals()[safe_class_name] = user_class

3. 关键细节说明 #

• CSV读取适配：你的示例CSV是制表符分隔的，所以用delimiter="\t"；如果是逗号分隔，直接改成delimiter=","即可。
• 权重配置：每个动态生成的用户类的weight值就是CSV里的“需模拟的用户数”，Locust会按权重比例分配用户数。
• 报告聚合优化：self.client.get(url, name=url.split('?')[0])这行代码可以把带有不同查询参数的同路径URL聚合到一起，让测试报告更清晰。
• 等待时间：between(1,3)表示每个用户在两次请求之间会等待1-3秒，模拟真实用户的操作间隔，你可以根据场景调整，比如改成between(0.5,2)。

4. 启动测试 #

假设你的脚本名为load_test.py，CSV里的用户数总和是8000（1000+5000+2000+1000），用以下命令启动：

locust -f load_test.py --headless -u 8000 -r 50

• -u：总模拟用户数，必须等于CSV里所有“需模拟的用户数”的总和。
• -r：每秒启动的用户数，可根据服务承受能力调整，避免瞬间压垮服务。

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