如何在Locust Python脚本中为循环调用生成递增唯一时间戳

我在Locust Python脚本开头定义了固定的时间戳变量：

from_date = "2023-07-01T08:00:00.000Z"
to_date = "2023-07-01T08:15:00.000Z"

后续在任务中循环使用这些固定值：

class SaveValuesUser(HttpUser):
    host = server_name
    # wait_time = between(10, 15)

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

        for i, (ts_id, ts_value) in enumerate(TS_IDs.items()):
            if i > 31:
                break

            data = get_data(ts_id, from_date, to_date, ts_value)
            data_list.append(data)
            
        json_data = json.dumps(data_list, indent=2)
        self.save_values(json_data)

    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)

我需要在for循环的每次迭代中使用递增、唯一且有效的时间戳，比如按天递增，但担心迭代到第30次时出现无效日期的问题，请问该如何实现？

方法1：基于初始时间递增固定间隔（单用户场景） #

利用Python的datetime模块处理时间计算，它会自动处理月末、闰年等边界情况，确保生成的时间戳始终有效。

1. 先导入所需模块：

from datetime import datetime, timedelta

2. 修改任务方法，在循环中动态计算递增的时间范围：

@task
def save_list_values(self):
    data_list = []
    # 将初始时间字符串转换为可计算的datetime对象
    base_from = datetime.fromisoformat(from_date.replace("Z", "+00:00"))
    base_to = datetime.fromisoformat(to_date.replace("Z", "+00:00"))
    # 定义每次迭代的时间增量，这里按天递增，可根据需求改为hours=15等
    time_increment = timedelta(days=1)

    for i, (ts_id, ts_value) in enumerate(TS_IDs.items()):
        if i > 31:
            break
        # 计算当前迭代的时间范围
        current_from = base_from + time_increment * i
        current_to = base_to + time_increment * i
        # 转换回要求的字符串格式（保留毫秒级精度）
        current_from_str = current_from.isoformat(timespec='milliseconds').replace("+00:00", "Z")
        current_to_str = current_to.isoformat(timespec='milliseconds').replace("+00:00", "Z")

        data = get_data(ts_id, current_from_str, current_to_str, ts_value)
        data_list.append(data)
    
    json_data = json.dumps(data_list, indent=2)
    self.save_values(json_data)

方法2：全局计数器+时间递增（多用户并发场景） #

如果Locust运行多用户并发，为避免不同用户的时间戳重复，可以用全局计数器配合锁来保证唯一性：

1. 初始化全局计数器和锁：

import threading

# 全局计数器，记录已使用的时间增量次数
timestamp_counter = 0
# 线程锁，保证多用户下计数器操作的原子性
counter_lock = threading.Lock()

2. 修改任务方法：

@task
def save_list_values(self):
    data_list = []
    base_from = datetime.fromisoformat(from_date.replace("Z", "+00:00"))
    base_to = datetime.fromisoformat(to_date.replace("Z", "+00:00"))
    time_increment = timedelta(days=1)

    # 抢占当前用户需要的计数器区间，避免多用户冲突
    global timestamp_counter
    with counter_lock:
        start_idx = timestamp_counter
        timestamp_counter += 32  # 循环最多执行32次（i从0到31）

    for i, (ts_id, ts_value) in enumerate(TS_IDs.items()):
        if i > 31:
            break
        # 基于全局计数器计算当前时间
        current_from = base_from + time_increment * (start_idx + i)
        current_to = base_to + time_increment * (start_idx + i)
        current_from_str = current_from.isoformat(timespec='milliseconds').replace("+00:00", "Z")
        current_to_str = current_to.isoformat(timespec='milliseconds').replace("+00:00", "Z")

        data = get_data(ts_id, current_from_str, current_to_str, ts_value)
        data_list.append(data)
    
    json_data = json.dumps(data_list, indent=2)
    self.save_values(json_data)

关键说明 #

• datetime.fromisoformat可以直接解析带时区的时间字符串，替换Z为+00:00是为了兼容Python的格式要求。
• timedelta会自动处理日期边界：比如从2024-02-29加1天会得到2024-03-01，完全不用担心无效日期问题。
• 如果需要更细粒度的唯一值，可将时间增量改为分钟/秒级，或者结合用户ID、迭代次数拼接时间戳。

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