1、引言

今天分享的这部分内容，应该算是Locust的进阶篇，

毕竟针对一般的性能自动化测试人员来说，

掌握小鱼写的前5章节的知识，就能足够应对大部分情况。

但是，针对有些需要进阶自己的技术，

那么，可以持续关注小鱼的博客，

让我们一起探索Locust，探索性能自动化。

那么，话不多说，我们开始今天的进阶篇，

自定义负载测试图形

2、定义

有些时候，默认的形状已经无法满足我们的特定要求，

那么这个时候，我们就需要完全自定义负载测试的图形形状。

而这并不难，就是通过设置 用户或者更更改用户数和产生率来实现的。

2.1 列举实例

例如：我们想自定义时间生成负载峰值或上下倾斜。如何实现呢？？

直接使用 LoadTestshape类，它可以完全的控制用户数和产生率。

是不是很Nice。

2.2 如何继承

在Locustfile文件中自定义一个继承LoadTestShape类的类，

如果有，则自动使用，

如果没有，则无法使用。

2.3 方法使用

1、在 此类中， 定义tick()方法：

该方法返回具有所有用户数和产生率的元组(无，则停止执行)。

Locust 大约每秒都会调用一次 tick()方法。

2、在该类用，访问 get_run_time()方法：

来检查测试运行的时间。

3、代码实战

3.1 时间峰值

1、背景

在此形状类别，将100为单位，增加用户数，然后运行10分钟后停止。

2、代码

# -*- coding: utf-8 -*-
"""
@ auth : carl_奕然
@ time : 2022-07-12
"""
from locust import LoadTestShape
class CustomShape(LoadTestShape):
#设置时限
time_limit = 600
#设置产生率
spawn_rate = 20
def tick(self):
'''
设置 tick()函数
并在tick()里面调用 get_run_time()方法
'''
#调用get_run_time()方法
run_time = self.get_run_time()
#做比较,运行时间在 10分钟之内，则继续执行
if run_time < self.time_limit:
#将用户数四舍五入到最接近的百分值
uesr_count =round(run_time,-2)
#返回user_count,spawn_rate这两个参数
return (user_count,spawn_rate)
return None

嗯，看着这个代码，是不是觉得没什么难度。

就短短的几行代码，这就实现了？？

嗯，是的。确实，

不仅定义这一个图形，还有好几个呢。

跟着小鱼，接着往下看。

3.2 双波形

代码示例

老规矩，直接上代码，在代码中解析：

# -*- coding: utf-8 -*-
"""
@ auth : carl_奕然
@ time : 2022-07-12
"""
import math
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
#设置task
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class DoubleWave(LoadTestShape):
    '''
    自定义一个双波形图形，
    模拟在某两个时间点的最高值
    参数解析:
        min_users ： 最小用户数
        peak_one_users ： 用户在第一个峰值
        peak_two_users ： 用户在第二个峰值
        time_limit ： 测试执行总时间
    '''
    # 最小用户数
    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）
             * 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

3.3 基于时间阶段

代码示例

同样，上代码，在代码中解析：

# -*- coding: utf-8 -*-
"""
@ auth : carl_奕然
@ time : 2022-07-12
"""
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class StagesShape(LoadTestShape):
    '''
    在不同的阶段 具有不同的用户数和 产生率的 图形形状
    参数解析:
        stages ：字典列表，每个字典都具有下列这些键值的阶段:
            duration -- 持续时间，  经过多少秒后，进入到下个阶段
            users -- 总用户数
            spawn_rate -- 产生率，即每秒启动/停止的用户数
            stop -- 可以在特定阶段停止测试的值
        stop_at_end -- 可以在所有阶段设置运行后停止
    '''
    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 10},
        {"duration": 100, "users": 50, "spawn_rate": 10},
        {"duration": 180, "users": 100, "spawn_rate": 10},
        {"duration": 220, "users": 30, "spawn_rate": 10},
        {"duration": 230, "users": 10, "spawn_rate": 10},
        {"duration": 240, "users": 1, "spawn_rate": 1}
    ]
    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stages['duration']:
                tick_data = (stage['users'],stage['spawn_rate'])
                return tick_data
        return None

3.4 逐步加载

代码示例

同样，上代码，在代码中解析：

# -*- coding: utf-8 -*-
"""
@ auth : carl_奕然
@ time : 2022-07-12
"""
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class StepLoadShaper(LoadTestShape):
    '''
    逐步加载实例
    参数解析：
        step_time -- 逐步加载时间
        step_load -- 用户每一步增加的量
        spawn_rate -- 用户在每一步的停止/启动 
        time_limit -- 时间限制
    '''
    setp_time = 30
    setp_load = 10
    spawn_rate = 10
    time_limit =  600
    def tick(self):
        run_time = self.get_run_time()
        if run_time > self.time_limit:
            return None
        current_step = math.floor(run_time /self.setp_time) +1
        return(current_step * self.setp_load,self.spawn_rate)

今天的内容，就是这么多，

总的来说，还算不难，虽然小鱼把这篇划分到进阶篇，

但是难度：★★★☆

我是Carl_奕然，

51Testing认证讲师

CSDN博客专家

关注我，带你学习更多更专业的测试开发知识。