技术教程 编程教程 Locust 任务类介绍 正文

Locust 任务类介绍

2023-05-29,

前言：

任务：简单的理解就是，你想要你脚本的虚拟用户去做哪些事，比如请求某一个接口，或者执行某一个事件
用户：可以理解为，执这个任务的实例主体，或者在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):

        # 单机器运行

        # 从机 --worker --master-host=192.168.0.56

        os.system('start locust -f FastHttpUser.py  --worker')

Locust 任务类介绍的相关教程结束。

《Locust 任务类介绍.doc》

下载本文的Word格式文档，以方便收藏与打印。

Copyright 2023. 昆居客 - IT编程,互联网信息技术文档大全!