使用Locust进行接口性能测试：Locust and TaskSet类详细分析（二）

左手的ㄟ右手 2023-10-14 13:52 20阅读 0赞

> Locust是一款开源的Python性能测试工具，它可以模拟大量并发用户对网站或者其他接口进行压力测

### 一、Locust类详细说明 ###

在Locust中，Locust类是整个负载测试工具的核心。它用于创建并发用户场景，模拟用户行为。示例：

from locust import Locust, TaskSet, task
    # 每一个Locust 类，就相当于jmeter中的每一个线程组。
    
    class MyUser(Locust):
        min_wait = 1000  # 最小等待时间（毫秒）
        max_wait = 5000  # 最大等待时间（毫秒）
            weight = 1                  # 权重，可以理解为jmeter中的线程组的权重。
    
    
        class MyTaskSet(TaskSet):
            @task
            def my_task(self):
                # 执行任务的逻辑，例如发送HTTP请求
                pass
    
        task_set = MyTaskSet

`MyUser`类，指定`min_wait`和`max_wait`属性，用于控制任务执行之间的随机等待时间范围。内部`MyTaskSet`类作为用户行为的集合。在`MyTaskSet`中，`@task`装饰器标装饰`my_task`让它成为一个执行任务。

### 二、TaskSet类详细说明 ###

TaskSet 类：实现了虚拟用户所执行任务的调度算法，包括:

1.  规划任务执行顺序（schedule\_task）
2.  挑选下一个任务（execute\_next\_task）
3.  执行任务（execute\_task）
4.  休眠等待（wait）
5.  中断控制（interrupt）

认清这些规则后，在`TaskSet`子类中可以采用非常简洁的方式来描述虚拟用户的业务测试场景，对虚拟用户的所有行为（任务）进行组织和描述，并可以对不同任务的权重进行配置。  
在TaskSet子类中定义任务信息时，可以采取两种方式:

*  @task装饰器
 *  tasks属性。

#### 2.1 @task装饰器 ####

`@task`装饰一个函数作为一个可执行任务。当Locust运行时，它将从TaskSet中随机选择任务，并以一定的频率执行这些任务。

from locust import Locust, TaskSet, task
    class MyUser(Locust):
        # ...
        class MyTaskSet(TaskSet):
            @task(1) # 设置任务执行频率，数值越大越高
            def task1(self):
                # 执行任务1
                pass
            @task(2) # 设置任务执行频率，数值越大越高
            def task2(self):
                # 执行任务2
                pass

在这个例子中，`MyTaskSet`类中定义了两个任务函数`task1`和`task2`，它们都被标记为可执行任务。Locust将随机选择并执行这两个任务。

#### 2.2 tasks属性 ####

`tasks`属性也可以设置频率

from locust import Locust, TaskSet, task
    
    class MyUser(Locust):
        # ...
    
        class MyTaskSet(TaskSet):
            tasks = {task1: 2, task2: 1} # 设置执行频率
              
            def task1(self):
                # 执行任务1
                pass
            
            def task2(self):
                # 执行任务2
                pass

`task1`的执行概率是`task2`的两倍。

#### 2.3 on\_start函数 ####

`on_start`用于任务开始执行任务前的预处理操作。例如，登录或初始化数据。它主要是使用了requests.Session，所以执行一次就可以让后续所有任务执行过程中都有登录态。

from locust import Locust, TaskSet, task
    
    class MyUser(Locust):
        # ...
    
        class MyTaskSet(TaskSet):
            @task(1)
            def my_task(self):
                self.client.get("/list")
    	        # 其他逻辑
                pass
    
            def on_start(self):
            # 在用户开始执行任务前执行的操作，例如登录
    	    self.client.post("/login")
                pass

`on_start`函数使用非常频繁。

#### 2.4 控制任务的执行顺序 ####

`@seq_task()`可以控制任务的执行顺序。  
按顺序执行任务：

from locust import Locust, TaskSequence, seq_task, task, between
    
    class MyUser(Locust):
        wait_time = between(1, 3)  # 设置任务执行的等待时间范围
    
        # ...
    
        class MyTaskSequence(TaskSequence):
            @seq_task(1)  # 顺序1
            def login(self):
                # 执行登录操作
                pass
            
            @seq_task(2)  # 顺序2
            def view_items(self):
                # 执行查看商品列表的操作
                pass
    
            @seq_task(3)  # 顺序3
            @task(3)  # 为该任务设置执行概率
            def add_to_cart(self):
                # 随机执行添加商品到购物车的操作，概率为3/4
                pass
    
            @seq_task(4)  # 顺序4
            def checkout(self):
                # 执行结账操作
                pass

#### 2.5 休眠等待（wait） ####

在任务之间插入休眠时间`wait_time`，可以模拟用户在执行任务时的等待行为，且Locust内置一些函数，返回`wait_time`方法

*  `between(min, max)` 生成指定范围内的随机等待时间, `min_wait`和`max_wait`分别表示等待时间的最小值和最大值（以毫秒为单位）
 *  `constant(constant_wait)` 生成固定的等待时间,`constant_wait`表示等待的时间（以毫秒为单位）
 *  `constant_pacing(pacing_interval)` 指定任务执行的固定间隔，比如：`constant_pacing(2)` 表示用户执行任务2秒，然后再继续下一个任务。

from locust import Locust, between, constant_pacing
    
    class MyUser(Locust):
        # 使用between函数，每个用户等待1到3秒之间的随机时间
        wait_time = between(1000, 3000)
    
    #  -----------------分割线------------------------------------
    class MyUser(Locust):
        # 使用constant_pacing属性，每个用户任务之间固定间隔2秒
        wait_time = constant_pacing(2)

#### 2.6 TaskSets 嵌套 ####

可以在一个TaskSet中嵌套其他TaskSet，实现更复杂的用户行为模拟。例如：

from locust import Locust, TaskSet, task
    
    class MyUser(Locust):
        # ...
    
        class SubTaskSet(TaskSet):
            @task(1)
            def sub_task1(self):
                # 执行子任务1
                pass
            
            @task(2)
            def sub_task2(self):
                # 执行子任务2
                pass
    
        class MyTaskSet(TaskSet):
            @task
            def task1(self):
                # 执行任务1
                pass
    
            @task
            def task2(self):
                # 执行任务2，并嵌套执行子任务集
                self.sub_task_set()
    
            tasks = {task1: 1, task2: 1}
    
            def sub_task_set(self):
                self.client.get("/otherurl")  
                self.SubTaskSet(self.locust).run()

在`MyTaskSet`中的`task2`任务中，我们嵌套执行了`SubTaskSet`。这样就可以实现一些非常复杂的业务流程的压测了，比如：用户打开网站首页，可能会直接离开，也可能会去流量商品，也可能加入商品到购物车下单等等场景一起模拟。

#### 2.7 中断控制(interrupt) ####

做压测的时候，有些场景下是要提前结束某些任务的，那么可以设置一个计数器，在计数到一定值后中断任务的执行，比上述的嵌套场景中也就刚刚好符合这点，嵌套子任务执行达到某一个阈值的时候，就中断回到父任务，这样就阻止了子任务一直执行的情况发生。

from locust import Locust, TaskSet, InterruptTaskSet, task, between
    
    
    class MyUser(Locust):
        wait_time = between(1, 3)  # 设置任务执行的等待时间范围
    
        class SubTaskSet(TaskSet):
            def no_start(self):
                # 计数
                self.counter = 0
    
            @task(1)
            def sub_task1(self):
                # 执行子任务1
                # 一些任务逻辑
                pass
    
            @task(2)
            def sub_task2(self):
                # 执行子任务2，达到计数后中断退出子任务
                if self.counter > 100:
                    self.interrupt()
                self.counter += 1
                # 一些任务逻辑
                pass
    
        class MyTaskSet(InterruptTaskSet):
    
            @task(1)
            def task1(self):
                # 执行任务1
                # 一些任务逻辑
                pass
    
            @task(2)
            def task2(self):
                "这里被中断后，后续不会再次触发执行"
                try:
                    # 执行任务2，中断控制
                    self.interrupt()
                except InterruptTaskSet as e:
                    # 中断任务后的处理
                    pass
    
            @task(3)
            def task3(self):
                # 执行任务3，并嵌套执行子任务集
                self.sub_task_set()
    
            @task(4)
            def task4(self):
                # 执行任务4
                # 一些任务逻辑
                pass
    
            def sub_task_set(self):
                # 执行子任务集合，并在子任务集合中使用中断控制
                self.SubTaskSet(self.locust).run()

子任务在每次执行任务时检查计数器的值，当计数器达到100时调用`interrupt`方法中断任务执行，返回顶层。

**最后：**下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取**【保证100%免费】**

![8b0171187dce405e878f274a8cdd9ec0.png][]

#### **软件测试面试文档** ####

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。

![a3e4226c935c4242804cbdf26b571bc6.png][]

![58df6cd288094ad8874e814f2477620b.gif][]

[8b0171187dce405e878f274a8cdd9ec0.png]: https://img-blog.csdnimg.cn/8b0171187dce405e878f274a8cdd9ec0.png
[a3e4226c935c4242804cbdf26b571bc6.png]: https://img-blog.csdnimg.cn/a3e4226c935c4242804cbdf26b571bc6.png
[58df6cd288094ad8874e814f2477620b.gif]: https://img-blog.csdnimg.cn/58df6cd288094ad8874e814f2477620b.gif