有时候默认的场景无法满足我们的要求时，这时后我们需要自定义负载策略。

比如我们要设置每一段时间启动100个用户运行，执行十分钟后再一次启动100个用户，总共运行10分钟，默认的场景是无法满足这样的要求的，我们可以使用LoadTestshape类，LoadTestshape类提供了几种负载测试策略

• 用法：

在脚本文件中定义一个继承LoadTestshape类的类，locust在启动时发现文件中有使用这个类会自动启动。

在该类中需要定义tick()方法，该方法返回用户数以及产生率的元组（如果没有返回这两个测试将停止），locust启动后每秒调用一次该函数。

在LoadTestshape类中可以使用get_run_time()方法来获取测试运行的时间，使用此方法可以用来控制压测的总时间。

1、时间峰值策略

class UserBehavior(TaskSet):
    @task(1)
    def test_01(self):
        param = {}
        with self.client.post('/api/product/list', json=param) as response:
            print("浏览首页商品列表")
​
    @task(1)
    def test_02(self):
        param = {}
        with self.client.post('/api/product/detail?id=1', json=param) as response:
            print("查看首页商品详情")
​
class CustomShape(LoadTestShape):
    # 设置时限
    time_limit = 300
    # 设置产生率
    spawn_rate = 20
​
    def tick(self):
        run_time = self.get_run_time()
        # 运行时间在 10分钟之内，则继续执行
        if run_time < self.time_limit:
            user_count = round(run_time, -2)
            # 返回user_count,spawn_rate这两个参数
            return (user_count, self.spawn_rate)
​
        return None
​
class WebsiteUser(HttpUser):
    host = 'http://www.baidu.com'
    tasks = [UserBehavior]
​
    wait_time = between(1, 2)
​
if __name__ == '__main__':
    os.system("locust -f locustfile3.py")

round(x,n),n的值可以是负数，表示在整数位部分四舍五入，如round(40,-2)=0, round(160,-2)=200，由此可得知

在（0~50s）时，user_count=0，

在（51s~150s）时，user_count=100；

在（151s~250s）时，user_count=200；

在（251s~300s）时，user_count=300；

查看运行结果：

2、时间阶段负载策略

class UserBehavior(TaskSet):
    @task(1)
    def test_01(self):
        with self.client.get('?scan') as response:
            print('浏览首页')
​
    @task(1)
    def test_02(self):
        with self.client.get('?item') as response:
            print('商品详情')
​
​
class CustomShape(LoadTestShape):
    stages=[
        {"time": 10, "users": 10, "spawn_rate": 10},
        {"time": 30, "users": 30, "spawn_rate": 10},
        {"time": 60, "users": 60, "spawn_rate": 10},
        {"time": 200, "users": 120, "spawn_rate": 10},
    ]
​
    def tick(self):
        run_time=self.get_run_time()
        for stage in self.stages:
            if run_time<stage['time']:
                tick_data=(stage['users'],stage['spawn_rate'])
                return tick_data
​
        return None
​
class WebsiteUser(HttpUser):
    host = 'http://www.baidu.com'
    tasks = [UserBehavior]
​
    wait_time = between(1, 2)
​
if __name__ == '__main__':
    os.system("locust -f locustfile4.py")

在不同的阶段 具有不同的用户数和 产生率的 图形形状

time -- 持续时间经过多少秒后，进入到下个阶段

users -- 总用户数

spawn_rate -- 产生率，即每秒启动/停止的用户数

查看运行结果：

3、逐步负载策略

class UserBehavior(TaskSet):
    @task(1)
    def test_01(self):
        with self.client.get('?scan') as response:
            print('浏览首页')
​
    @task(1)
    def test_02(self):
        with self.client.get('?item') as response:
            print('商品详情')
​
​
class MyCustomShape(LoadTestShape):
    setp_time = 30
    setp_load = 10
    spawn_rate = 10
    time_limit = 300
​
    def tick(self):
        run_time = self.get_run_time()
        if run_time<self.time_limit:
            current_step = math.floor(run_time / self.setp_time) + 1
            return (current_step * self.setp_load, self.spawn_rate)
        return None
​
class WebsiteUser(HttpUser):
    host = 'http://www.baidu.com'
    tasks = [UserBehavior]
​
    wait_time = between(1, 2)
​
if __name__ == '__main__':
    os.system("locust -f locustfile5.py")

step_time -- 逐步加载时间长度 step_load -- 用户每一步增加的量 spawn_rate -- 用户在每一步的停止/启动的多少用户数 time_limit -- 时间限制压测的执行时长

逐步负载策略每隔30秒新增启动10个用户，运行结果如下：

4、双波形

class UserBehavior(TaskSet):
    @task(1)
    def test_01(self):
        with self.client.get('?scan') as response:
            print('浏览首页')
​
    @task(1)
    def test_02(self):
        with self.client.get('?item') as response:
            print('商品详情')
​
​
class DoubleWave(LoadTestShape):
    # 最小用户数
    min_users = 20
​
    # 第一个峰值的用户数
    peak_one_users = 60
​
    # 第二个峰值的用户数
    peak_two_users = 40
​
    # 测试执行时间
    time_limit = 600
​
    def tick(self):
​
        # 将get_run_time 四舍五入
        run_time = round(self.get_run_time())
​
        if run_time < self.time_limit:
​
            user_count = (
                (self.peak_one_users - self.min_users)
                * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 5) ** 2)
                + (self.peak_two_users - self.min_users)
                * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 10) ** 2)
                + self.min_users
​
            )
            return (round(user_count), round(user_count))
        else:
            return None
​
​
class WebsiteUser(HttpUser):
    host = 'http://www.baidu.com'
    tasks = [UserBehavior]
​
    wait_time = between(1, 2)
​
if __name__ == '__main__':
    os.system("locust -f locustfile6.py")

min_users ： 最小用户数

peak_one_users ： 用户在第一个峰值

peak_two_users ： 用户在第二个峰值

time_limit ： 测试执行总时间

自定义一个双波形图形， 模拟在某两个时间点的最高值，运行结果如下：