自定义负载策略

在Locust中（直接压测脚本文件中）可直接定义一个继承LoadTestshape类的类，有则自动启动。

1. 在该类中定义tick()方法，返回用户数user_count和产生率spawn_rate的元组，如果返回None则停止测试
2. _通过_get_run_time()方法获取测试运行时间，使用此方法控制压测时间

基于时间峰值策略

# 基于时间峰值策略： 每秒生成10个用户，持续时间60s
class TimePeadShape(LoadTestShape):
    # time_limit设置整个压测过程为60秒
    time_limit = 60
    # 设置产生率一次启动10个用户
    spawn_rate = 10
    def tick(self):
        """
        设置 tick()函数
        并在tick()里面调用 get_run_time()方法
        """
        # 调用get_run_time()方法获取压测执行的时间
        run_time = self.get_run_time()
        # 运行时间在 time_limit之内，则继续执行
        print('run_time', run_time)
        if run_time < self.time_limit:
            # 每10秒钟增加10个用户
            user_count = round(run_time, -1)
            return user_count, self.spawn_rate
        return None

基于步骤负载策略

# 基于步骤负载策略：每5秒增加10个用户，持续1分钟
class StepShape(LoadTestShape):
    """
        step_time -- 步骤时间间隔
        step_load -- 用户每一步增加数量
        spawn_rate -- 用户每一步每秒停止/启动数量
        time_limit -- 时间限制，以秒为单位
    """
    step_time = 5
    step_load = 10
    spawn_rate = 10
    time_limit = 60
    def tick(self):
        # 调用get_run_time()方法获取压测执行的时间
        run_time = self.get_run_time()
        print('run_time', run_time)
        # 运行时间在 time_limit之内，则继续执行
        if run_time > self.time_limit:
            return None
        # 判断当前负载用户数
        current_step = math.floor(run_time / self.step_time) + 1
        print('current_step', current_step)
        return current_step * self.step_load, self.spawn_rate

基于时间阶段负载策略

�

# 基于时间段负载策略：前10s和10-20s用户数为10；20-50s用户数为50；50-80s用户数为100；80s后用户数为30
class TimeStageShape(LoadTestShape):
    """
        duration -- 多少秒后进入下一个阶段
        users -- 用户数
        spawn_rate -- 每秒要启动/停止的用户数
    """
    stages = [
        {"duration": 10, "users": 10, "spawn_rate": 10},
        {"duration": 20, "users": 50, "spawn_rate": 10},
        {"duration": 50, "users": 100, "spawn_rate": 10},
        {"duration": 80, "users": 30, "spawn_rate": 10}
    ]
    def tick(self):
        # 调用get_run_time()方法获取压测执行的时间
        run_time = self.get_run_time()
        for stage in self.stages:
            # 判断运行时间在不同阶段负载不同用户数量
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data
        return None