单个Locust测试中运行读写任务及配置任务权重的问题咨询

报错信息 #

N52820/ERROR/locust.user.task: Population must be a sequence. For dicts or sets, use sorted(d).

原测试代码 #

from locust import HttpUser, between, task
from locust.contrib.fasthttp import FastHttpUser

import random
import json
import csv
from datetime import datetime, timedelta

server_name = "https://rknwvb.com:7072"
save_from_date = "2023-07-05T08:00:00.000Z"
save_to_date = "2023-07-05T09:00:00.000Z"


headers = {'X-API-KEY': 'AKjCg9hTcYQ=', 'Content-Type': 'application/json'}

LOAD_TSIDs = {
'15016',
'83904',
'87371'
}

SAVE_TS_IDs = {
'14881': 30,
'15016': 30,
'87371': 30
}


def set_value_save_values():
    ts_value_random = random.randrange(10000)
    print("random_value:", ts_value_random)
    return ts_value_random


def get_data_save_values(ts_id, from_date, to_date, ts_value):
    myjson = {
        "id": int(ts_id),
        "values": [
            {
                "from": from_date,
                "to": to_date,
                "value": ts_value
            }
        ]
    }
    return myjson


def set_from_date_load_values():
    min_tid = datetime.now()
    nytid = min_tid.replace(year=2022, month=1, day=1, hour=random.randint(0, 23), minute=0, 
second=0, microsecond=0)
    from_date = nytid.strftime("%Y-%m-%dT%H:%M:00.000Z")
    print("from_date:", from_date)
    return from_date


def set_to_date_load_values():
    min_tid = datetime.now()
    nytid = min_tid.replace(year=2022, month=1, day=2, hour=random.randint(0, 23), minute=0, 
second=0, microsecond=0)
    to_date = nytid.strftime("%Y-%m-%dT%H:%M:00.000Z")
    print("to_date:", to_date)
    return to_date


class SaveAndLoadValues(FastHttpUser):
    host = server_name

    def _run_read_ts(self, series_list, resolution, start, end):
        LOAD_TSIDs = ",".join(series_list)
        resp = self.client.get(
            f'/api/loadValues?tsIds={LOAD_TSIDs}&resolution={resolution}&startUtc={set_from_date_load_values()}&endUtc={set_to_date_load_values()}',
            headers={'X-API-KEY': 'AKjCg9hTcYQ='}
        )
        print("Response status code:", resp.status_code)

    def save_values(self, json_data):
        print(type(json_data))
        print(json_data)
        # Make the PUT request with authentication:
        response = self.client.put("/api/SaveValues", data=json_data, headers=headers)
        # Check the response:
        if response.status_code == 200:
            print("SaveValues successful!")
            print("Response:", response.json())
        else:
            print("SaveValues failed.")
            print("Response:", response.text)

    @task(1)
    def test_get_ts_1(self):
        self._run_read_ts(random.sample(list(SAVE_TS_IDs), 1), 'PT15M', set_from_date_load_values(),
                        set_to_date_load_values())

    @task(2)
    def save_list_values(self):
        data_list = []

        # 修正样本数量，避免超过字典键的总数
        for i, (ts_id, ts_value) in enumerate(random.sample(list(SAVE_TS_IDs.items()), random.randint(1,3))):
            data = get_data_save_values(ts_id, save_from_date, save_to_date, set_value_save_values())
            data_list.append(data)

        json_data = json.dumps(data_list, indent=2)
        self.save_values(json_data)

问题解答 #

1. 单个Locust测试中同时运行读写任务的方式是否正确？ #

完全正确。Locust的核心设计就是支持在同一个User类中定义多个@task装饰的方法，模拟用户的多样化操作行为。这种混合读写任务的模式，能够精准还原真实业务场景中用户既有读操作又有写操作的情况，是Locust性能测试的常规用法。

2. 是否可以设置任务相对权重，实现1次读操作对应10次写操作的模式？ #

可以直接通过调整@task装饰器的权重参数实现。只需给读任务设置@task(1)，写任务设置@task(10)，Locust在调度任务时会严格按照权重比例分配执行次数，长期统计下来就能达到平均1次读对应10次写的效果。

额外报错修复说明 #

• 报错根源：random.sample()要求第一个参数是序列类型（如list），你直接传入了字典/集合，需转成list，比如random.sample(list(SAVE_TS_IDs), 1)
• 修正样本数量：SAVE_TS_IDs仅包含3个键，不能使用random.sample(..., 100)，会触发样本数量超过总体数量的错误，需调整为不超过3的数值
• 修正缩进错误：save_values和save_list_values原本被错误嵌套在其他方法中，需调整为类的同级方法
• 修正URL拼接：原GET请求的URL存在换行语法错误，需调整为合法的字符串拼接格式

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