Locust 任务类介绍

Posted 2023-03-29 肮脏都视而不见，出淤泥不染的莲

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Locust 任务类介绍相关的知识，希望对你有一定的参考价值。

前言：

任务：简单的理解就是，你想要你脚本的虚拟用户去做哪些事，比如请求某一个接口，或者执行某一个事件
用户：可以理解为，执这个任务的实例主体，或者在locust 中，也可以认为是一群蝗虫

一、TaskSet 类

TaskSet是定义用户将执行的一组任务的类，然后通过虚拟用户调用并执行这一组任务

1、单个任务集合

# -*- coding: utf-8 -*-
from locust import TaskSet, task, User
import os

# 单个任务集合
class Task_1(TaskSet):

    @task
    def task_a(self):
        print(\'打开冰箱门\')

    @task
    def task_b(self):
        print(\'把大象装进冰箱\')

    @task
    def task_c(self):
        print(\'关上冰箱门\')

class task_conf(User):
    tasks = [Task_1]

if __name__ == \'__main__\':
    os.system(\'locust -f TaskSet_single.py --web-host "127.0.0.1"\')

2、多个任务集合

# -*- coding: utf-8 -*-
from locust import TaskSet, task, User
import os


\'\'\' 
TaskSet类定义了每个用户的任务集合，
注意：如果存在多个任务集合，需要在每个任务集合中调用interrupt()方法，
否则用户一旦进入到这个用户，则无法跳出这个任务集合（会一直执行这个任务集）
\'\'\'

class Task_1(TaskSet):

    @task
    def task_a(self):
        print(\'打开冰箱门\')

    @task
    def task_b(self):
        print(\'把大象装进冰箱\')

    @task
    def task_c(self):
        print(\'关上冰箱门\')

    # 调用interrupt() 非常重要，否则用户将难以自拔
    @task
    def logout(self):
        self.interrupt()

class Task_2(TaskSet):

    @task
    def task_1(self):
        print(\'窗前明月光,\')

    @task
    def task_2(self):
        print(\'疑是地上霜.\')

    @task
    def task_3(self):
        print(\'举头望明月,\')

    @task
    def task_4(self):
        print(\'低头思故乡.\')

    @task
    def logout(self):
        self.interrupt()

class task_conf(User):
    tasks = [Task_1, Task_2]

if __name__ == \'__main__\':
    os.system(\'locust -f TaskSet_multiple.py --web-host "127.0.0.1"\')

二、User 类

User 执行测试任务的主体类，用户的行为由其任务定义

# -*- coding: utf-8 -*-
from locust import task, User
import os

\'\'\' 
通过User 定义任务
\'\'\'

class Task_1(User):

    @task
    def task_a(self):
        print(\'打开冰箱门\')

    @task
    def task_b(self):
        print(\'把大象装进冰箱\')

    @task
    def task_c(self):
        print(\'关上冰箱门\')

if __name__ == \'__main__\':
    os.system(\'locust -f User.py --web-host "127.0.0.1"\')

三、任务属性

测试开始时，locust将为每一个模拟用户创建一个User类的实例，每个实例都在他们自己的微线程中执行，当这些用户运行时，开始选中要执行的任务，休眠一段时间，然后选一个新的任务，继续执行，以此类推。

这里我们为用户添加任务的方式有两种：

使用@task 装饰器
设置tasks属性

1、@task

通过@task 定义用户任务任务 ,同时还可以通过@task(int) 设置每个任务的权重

# -*- coding: utf-8 -*-
from locust import task,User
import os

# 通过@task 定义用户任务 任务 ,同时还可以通过@task(int) 设置每个任务的权重
class task_2(User):

    @task(2)
    def task_a(self):
        print(\'打开冰箱门\')

    @task(2)
    def task_b(self):
        print(\'把大象装进冰箱\')

    @task(1)
    def task_c(self):
        print(\'关上冰箱门\')


if __name__ == \'__main__\':
    os.system(\'locust -f User_task.py --web-host="127.0.0.1"\')

2、tasks

# -*- coding: utf-8 -*-
from locust import task,User
import os

\'\'\' 
继承TaskSet，不按顺序执行任务
\'\'\'

# 通过tasks 定义测试任务
class task_1(User):

    def task_a(self):
        print(\'打开冰箱门\')

    def task_b(self):
        print(\'把大象装进冰箱\')

    def task_c(self):
        print(\'关上冰箱门\')

    # 通过tasks 定义测试任务 并设置对应任务执行权重
    tasks = [task_a, task_b, task_c]
    # tasks = task1:1,task2:2,task3:2

if __name__ == \'__main__\':
    os.system(\'locust -f User_tasks.py --web-host="127.0.0.1"\')

3、@tag任务标记

可以通过标记的方式，执行负载测试时，让虚拟用户只执行指定的测试任务

# -*- coding: utf-8 -*-
from locust import User, tag

\'\'\'
设置tag，用于设置用户 执行指定标记的任务
\'\'\'
class myUser(User):

    @tag(\'回归\')
    def task_1(self):
        print(\'回归任务\')

    @tag(\'流程\')
    def task_2(self):
        print(\'流程任务\')

    @tag(\'回归\', \'流程\')
    def task_3(self):
        print(\'回归任务 + 流程任务\')

    tasks = [task_1, task_2, task_3]

if __name__ == \'__main__\':
    import os
    os.system(\'locust -f User_tag.py --tags 回归 --web-host="127.0.0.1"\')

四、User 类通用属性

Locust 将为每个用户生成一个 User 类的实例。用户类可以定义一些通用属性

1、wait_time属性

用于模拟用户每次执行任务后的延迟时间（等同思考时间），如果未指定，则下一个任务将在完成后立即执行

constant(x)　　固定时间
between(x,y) 　最小值和最大值之间的随机时间
constant_packing(x) 每x秒最多运行一次任务
constant_throughtput(x) 每秒最多运行x次

# -*- coding: utf-8 -*-
from locust import constant, between, task,User
import os

\'\'\' 
wait_time = constant() 等待一个固定的时间
wait_time = between() 等待一个最小值和最大值之间的随机时间 不含最大时间
\'\'\'

class Task_1(User):

    @task
    def task_a(self):
        print(\'打开冰箱门\')

    @task
    def task_b(self):
        print(\'把大象装进冰箱\')

    @task
    def task_c(self):
        print(\'关上冰箱门\')

    # 用户执行每个任务，会固定等待5s
    wait_time = constant(5)

    # 用户执行每个任务，会固定等待[1,5) 秒
    # wait_time = between(1, 5)

if __name__ == \'__main__\':
    os.system(\'locust -f User_wttime.py --web-host "127.0.0.1"\')

2、weight权重属性

用于设置特定类型的用户的权重值，在未设置的情况下，locust将为每个用户类生产相同数量的用户实例

比如app用户是web用户的3倍

# -*- coding: utf-8 -*-
from locust import User

# App用户数量是Web用户数量的3倍
class appUser(User):
    weight = 3

    def task_a(self):
        print(\'App任务1\')

    def task_b(self):
        print(\'App任务2\')

    tasks = [task_a, task_b]

class webUser(User):
    weight = 1

    def task_A(self):
        print(\'Web任务1\')

    def task_B(self):
        print(\'Web任务2\')

    tasks = [task_A, task_B]

if __name__ == \'__main__\':
    import os
    os.system(\'locust -f User_weight.py --web-host="127.0.0.1"\')

3、host主机属性

用于加载主机的url 前缀。也可以在 webui中指定，或者命令参数中指定--host

# -*- coding: utf-8 -*-
from locust import task, HttpUser
import os

\'\'\'
主机属性 host 
\'\'\'

class task_s(HttpUser):

    host = \'我爱你\'
    @task
    def task_1(self):
        url = \'中国\'
        res = self.client.get(url=url, verify=False)
        print(res.json())

if __name__ == \'__main__\':
    os.system(\'locust -f User_host.py --web-host="127.0.0.1"\')

4、前后置on_start和on_stop方法

每个虚拟用户执行前。会执行一次该方法（在一次测试结束后，只会执行一次）

# -*- coding: utf-8 -*-
from locust import User, constant

\'\'\'
每个虚拟用户 执行前
会执行一次该方法（在一次测试结束后，只会执行一次）
\'\'\'

class myuser(User):
    wait_time = constant(2)

    def on_start(self):
        print(\'登录\')

    def on_stop(self):
        print(\'登出\')

    def task_1(self):
        print(\'访问主页数据\')

    tasks = [task_1]

if __name__ == \'__main__\':
    import os

    os.system(\'locust -f User_start_stop.py --web-host="127.0.0.1"\')

五、HttpUser 类

HttpUser是最常用的 User ，它通过client 发送Http请求,client是httpsession的一个实例（requests模块的session），故在locust中，通过client 发送的请求，具有天然的保持会话能力

1、发送http请求

# -*- coding: utf-8 -*-
from locust import HttpUser, task
import os

class TaskA(HttpUser):
    #登录
    def on_start(self):
        url = \'https://www.lit/login\'
        data = 
                \'login_user\': 51522,
                \'login_password\': \'pwd666\',
        
        res = self.client.post(url=url, data=data, verify=False)
   #登录成功后退出登录
    @task
    def test_loginout(self):
        url = \'https://www.lit/loginout\'
        res = self.client.post(url=url)
        print(res.text)

if __name__ == \'__main__\':
    os.system(\'locust -f HttpUser.py --web-host="127.0.0.1"\')

2、响应检查

HttpUser 支持设置响应检查点。使用catch_response参数、with语句和对response.failure() 标记为失败

# -*- coding: utf-8 -*-
from locust import HttpUser, task
import os
from json import JSONDecodeError

class task_s(HttpUser):
    @task
    def task_1(self):
        get_url = \'http://lit/loginout\'
        with self.client.post(url=get_url, catch_response=True)as response:
            try:
                # 获取响应内容的message 字段值
                status = response.json()[\'message\']
                # 获取返回信息
                # 如果message为“退出登录成功”则为成功，否则为失败
                if status == "退出登录成功":
                    response.success()
                else:
                    response.failure(\'非法的参数\')
            except JSONDecodeError:
                response.failure(\'Response could not be decoded as JSON\')

if __name__ == \'__main__\':
    os.system(\'locust -f HttpUser_catch.py --web-host="127.0.0.1"\')

六、FastHttpUser 类

Locust 的默认 HTTP 客户端使用python-requests，而requests库是一个同步阻塞库（每个请求发送完成之后都在等待响应的回来，这在高并发场景下是致命的）

正因为如此，Locust 还附带FastHttpUser使用geventhttpclient代替。它提供了一个非常相似的 API，并且使用的 CPU 时间显着减少，通常情况下将给定硬件上每秒的最大请求数增加了 5 到 6 倍

官方文档 FastHttpUser 和 HttpUser 比较，使用 FastHttpUsers 的测试将能够在每个核心每秒执行接近 5000 个请求，使用 HttpUser 大约 850 个

使用上：只需继承 FastHttpUser 就可以了，其他和 HttpUser 没有什么太大的差别

# -*- coding: utf-8 -*-
from locust import task, FastHttpUser
import os

# 进程数
WORKERS = int(1)

# URL
base_url = "http://11.12.0.13"
access_url = "/lit/number"

url = base_url + access_url

class task_ts(FastHttpUser):
    @task
    def task_api_1(self):
        # self.client.post
        res = self.client.get(url=url, verify=False)
        # 使用assert断言请求是否正确
        try:
            assert res.status_code == 200
            print(\'断言成功！\')
            res = res.json()
            print(res)
        except Exception as err:
            print(\'断言失败！\')

if __name__ == \'__main__\':
    os.system(\'start locust -f FastHttpUser.py --master --host=\' + base_url)
    for i in range(WORKERS):
        os.system(\'start locust -f FastHttpUser.py  --worker\')

以上是关于Locust 任务类介绍的主要内容，如果未能解决你的问题，请参考以下文章

locust使用经验---每个任务使用新的链接以及处理大量TIME_WAIT连接

2Locust压力测试实战

locust性能测试简谈

性能测试工具Locust | iCourt技术

动态创建locust任务

python服务器性能测试工具locust使用指南