locust性能测试：（十）自定义负载策略

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

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

  • 用法：

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

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

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

1、 时间峰值策略：

    from locust import HttpUser,TaskSet,between,task,LoadTestShape
    import os,sys
    curPath = os.path.abspath(os.path.dirname(__file__))
    rootPath = os.path.split(curPath)[0]
    PathProject = os.path.split(rootPath)[0]
    sys.path.append(rootPath)
    sys.path.append(PathProject)
    
    # 首页
    class UserBehavior(TaskSet):
    
        @task(1)
        def test_01(self):
            """
            浏览首页商品列表
            :return:
            """
            param = {}
            with self.client.post('/api/product/list', json=param) as response:
                print("浏览首页商品列表")
    
        @task(1)
        def test_02(self):
            """
            查看首页商品详情
            :return:
            """
            param = {}
            with self.client.post('/api/product/detail?id=1', json=param) as response:
                print("查看首页商品详情")
    
    
    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 (uesr_count, self.spawn_rate)
    
            return None
    
    class WebsiteUser(HttpUser):
        host = 'http://127.0.0.1'
        tasks = [UserBehavior]
    
        wait_time = between(1, 2)
    
    if __name__ == '__main__':
        os.system("locust -f ccc.py")

查看运行结果：

image.png

每运行一分钟启动100个用户，总运行时间为10分钟

2、时间阶段负载策略

from locust import HttpUser,TaskSet,between,task,LoadTestShape
import os,sys
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
PathProject = os.path.split(rootPath)[0]
sys.path.append(rootPath)
sys.path.append(PathProject)
import math
# 首页
class UserBehavior(TaskSet):

    @task(1)
    def test_01(self):
        """
        浏览首页商品列表
        :return:
        """
        param = {}
        with self.client.post('/api/product/list', json=param) as response:
            print("浏览首页商品列表")

    @task(1)
    def test_02(self):
        """
        查看首页商品详情
        :return:
        """
        param = {}
        with self.client.post('/api/product/detail?id=1', json=param) as response:
            print("查看首页商品详情")



class MyCustomShape(LoadTestShape):
    """
        ps:在不同的阶段 具有不同的用户数和 产生率的 图形形状
        time -- 持续时间经过多少秒后，进入到下个阶段
        users -- 总用户数
        spawn_rate -- 产生率，即每秒启动/停止的用户数
    """
    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://127.0.0.1'
    tasks = [UserBehavior]

    wait_time = between(1, 2)

if __name__ == '__main__':
    os.system("locust -f ccc.py")

查看运行结果：

image.png

3、逐步负载策略

from locust import HttpUser,TaskSet,between,task,LoadTestShape
import os,sys
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
PathProject = os.path.split(rootPath)[0]
sys.path.append(rootPath)
sys.path.append(PathProject)
import math
# 首页
class UserBehavior(TaskSet):

    @task(1)
    def test_01(self):
        """
        浏览首页商品列表
        :return:
        """
        param = {}
        with self.client.post('/api/product/list', json=param) as response:
            print("浏览首页商品列表")

    @task(1)
    def test_02(self):
        """
        查看首页商品详情
        :return:
        """
        param = {}
        with self.client.post('/api/product/detail?id=1', json=param) as response:
            print("查看首页商品详情")


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

    # 逐步负载策略每隔30秒新增启动10个用户
    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:
            return None
        current_step = math.floor(run_time /self.setp_time) +1
        return(current_step * self.setp_load,self.spawn_rate)


class WebsiteUser(HttpUser):
    host = 'http://127.0.0.1'
    tasks = [UserBehavior]

    wait_time = between(1, 2)

if __name__ == '__main__':
    os.system("locust -f ccc.py")

运行结果：

image.png

4、双波形

from locust import HttpUser,TaskSet,between,task,LoadTestShape
import os,sys
curPath = os.path.abspath(os.path.dirname(__file__))
rootPath = os.path.split(curPath)[0]
PathProject = os.path.split(rootPath)[0]
sys.path.append(rootPath)
sys.path.append(PathProject)
import math
# 首页
class UserBehavior(TaskSet):

    @task(1)
    def test_01(self):
        """
        浏览首页商品列表
        :return:
        """
        param = {}
        with self.client.post('/api/product/list', json=param) as response:
            print("浏览首页商品列表")

    @task(1)
    def test_02(self):
        """
        查看首页商品详情
        :return:
        """
        param = {}
        with self.client.post('/api/product/detail?id=1', json=param) as response:
            print("查看首页商品详情")


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)
            + (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://127.0.0.1'
    tasks = [UserBehavior]

    wait_time = between(1, 2)

if __name__ == '__main__':
    os.system("locust -f ccc.py")

运行结果：

image.png
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容