本文共 3218 字，大约阅读时间需要 10 分钟。

目录

前言

• 有时候我们需要一个完全定制的负载测试，而这并不能通过简单地设置或改变用户数量和刷出率而实现。例如，可能希望在自定义时间生成一个负载尖峰或上升或下降。通过使用LoadTestShape类，您可以在任何时候完全控制用户计数和生成速率。

基于时间峰值策略

• 每秒生成10个用户，持续时间60s

# -*- coding: utf-8 -*-# @Time    : 2021/5/2# @Author  : 大海import osfrom locust import LoadTestShape, task, HttpUser, constantclass MyUser(HttpUser):    # 请求间隔时间    wait_time = constant(1)    host = 'https://www.baidu.com'    @task    def my_task(self):        self.client.get('/')class MyCustomShape(LoadTestShape):    # 设置压测持续时长，单位秒    time_limit = 60    # 每秒启动/停止用户数    spawn_rate = 10    def tick(self):        """        返回一个元组，包含两值：            user_count -- 总用户数            spawn_rate -- 每秒启动/停止用户数        返回None时，停止负载测试        """        # 获取压测执行的时间        run_time = self.get_run_time()        # 运行时长在压测最大时长内，则继续执行        if run_time < self.time_limit:            user_count = round(run_time, -1)            return user_count, self.spawn_rate        return Noneif __name__ == '__main__':    file_path = os.path.abspath(__file__)    os.system(f'locust -f {file_path} --web-host=127.0.0.1')

基于步骤负载策略

• 每30秒增加10个用户，持续10分钟

# -*- coding: utf-8 -*-# @Time    : 2021/5/2# @Author  : 大海import mathimport osfrom locust import HttpUser, TaskSet, task, constantfrom locust import LoadTestShapeclass UserTasks(TaskSet):    @task    def my_task(self):        self.client.get("/")class WebsiteUser(HttpUser):    wait_time = constant(0.5)    tasks = [UserTasks]    host = 'https://www.baidu.com'class StepShape(LoadTestShape):    """        step_time -- 步骤之间的时间        step_load -- 用户在每一步增加数量        spawn_rate -- 用户在每一步式每秒停止/启动数量        time_limit -- 时间限制，以秒为单位    """    step_time = 30    step_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.step_time) + 1        return current_step * self.step_load, self.spawn_rateif __name__ == '__main__':    file_path = os.path.abspath(__file__)    os.system(f'locust -f {file_path} --web-host=127.0.0.1')

基于时间阶段负载策略

• 前10s和10-20s用户数为10；20-50s用户数为50；50-80s用户数为100；80s后用户数为30

# -*- coding: utf-8 -*-# @Time    : 2021/5/2# @Author  : 大海import osfrom locust import HttpUser, TaskSet, task, constantfrom locust import LoadTestShapeclass UserTasks(TaskSet):    @task    def my_task(self):        self.client.get("/")class WebsiteUser(HttpUser):    wait_time = constant(0.5)    tasks = [UserTasks]    host = 'https://baidu.com'class TimeShape(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):        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 Noneif __name__ == '__main__':    file_path = os.path.abspath(__file__)    os.system(f'locust -f {file_path} --web-host=127.0.0.1')