locust 手册

赖荣生 / 2024-03-20 / 原文

wait_time 等待时间
- 1. between(min_wait, max_wait)
- 2. constant(wait_time)
- 3. constant_pacing(wait_time)
- 4. Constant_throughput(x)
- 5. 自定义 wait_time
weight 权重、fixed_count 固定次数
- 1. 如果文件中存在多个类，并且命令行中没有指定运行那个类，locust将从现有类中随机选择
- 2. 如果你想让 WebUser 执行得更频繁，你可以 weight（权重）属性。例如，WebUser用户是MobileUser用户的三倍
- 3. 也可以设置 fixed_count 属性。在这种情况下，权重属性将被忽略，并且将生指定数量的用户数。这些用户首先产生。
host 主机属性
- 1. host 属性是 URL 的前缀。它会自动添加到请求中。
- 2. 也可以在命令行中增加 --host 参数实现
- 3. 也可以在 Web UI 中输入
task 任务、@task 装饰器、tasks 任务属性
- 1. 装饰器方式，也就是在任务上添加 @task
- 2. 任务属性，也就是哪些是任务加到 tasks = [] 中
- 3. @task 可选权重参数，用于指定任务的执行比率。
@tag 标签装饰器
on_start 类开始运行前、on_stop 类结束运行前
test_start 压测开始前、test_stop 压测试结束前
init 启动 locust 时触发的
HttpSession
- 1. 隐式重用 cookie
- 2. 单次修改 sesion
- 3. 修改 sesion
Http
- get 请求
  - 1. 无参 get 请求
  - 2. 带参 get 请求
- post 请求
- 参数

wait_time 等待时间

用于确定模拟用户在执行任务之间将等待多长时间

1. between(min_wait, max_wait)

该函数将在 min_wait 和 max_wait 之间返回一个随机数，此例，在每个任务之间随机等待 0.1s 到 1s

wait_time = between(0.1, 1)

2. constant(wait_time)

该函数只返回 wait_time 参数指定的数字，此例，在每个任务之间固定等待 3s

wait_time = constant(3)

3. constant_pacing(wait_time)

确保任务每 X 秒运行一次，此例，每秒执行一次任务，无轮上次任务是否完成

wait_time = constant_pacing(1)

4. Constant_throughput(x)

指定每秒执行任务请求数。此例，每秒执行两次请求。

wait_time = Constant_throughput(2)

5. 自定义 wait_time

当然也支持自定义等待时间。
定义一个 wait_time 方法，每次调用次方法都返回变量（last_wait_time）加 1 的数
也就是第一次等待 1s，第二次等待 2s，依次类推

class MyUser(User):
    last_wait_time = 0

    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

weight 权重、fixed_count 固定次数

1. 如果文件中存在多个类，并且命令行中没有指定运行那个类，locust将从现有类中随机选择

当然你也可以使用如下命令行指定那个类

locust -f aa.py WebUser MobileUser

2. 如果你想让 WebUser 执行得更频繁，你可以 weight（权重）属性。例如，WebUser用户是MobileUser用户的三倍

3. 也可以设置 fixed_count 属性。在这种情况下，权重属性将被忽略，并且将生指定数量的用户数。这些用户首先产生。

下例中，将仅生成一个 AdminUser 实例，而与用户总数无关。这样无论你设置并发数是多少，就只会有一个 AdminUser

from locust import HttpUser, task, constant , constant_pacing

class AdminUser(HttpUser):
    fixed_count = 1
    wait_time = constant_pacing(1)
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        self.client.get("/p/1.html")

class MobileUser(HttpUser):
    weight = 10
    wait_time = constant_pacing(1)
    host = 'http://127.0.0.1:8080'

    @task(70)
    def task_3(self):
        self.client.get("/p/3.html")

host 主机属性

1. host 属性是 URL 的前缀。它会自动添加到请求中。

下例中，会访问这个地址 http://127.0.0.1:8080/p/1.html

from locust import HttpUser, task

class AdminUser(HttpUser):
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        self.client.get("/p/1.html")

2. 也可以在命令行中增加 --host 参数实现

locust -f a1.py --host http://127.0.0.1:8080

3. 也可以在 Web UI 中输入

task 任务、@task 装饰器、tasks 任务属性

User 类可以使用 @task 装饰器将声明任务，但也可以使用 tasks 属性指定任务。@task 装饰器、tasks 属性选一个就好，建议使用 @task 装饰器

1. 装饰器方式，也就是在任务上添加 @task

from locust import User, task, constant

class MyUser(User):
    wait_time = constant(1)

    @task
    def my_task(self):
        print("User instance (%r) executing my_task" % self)

2. 任务属性，也就是哪些是任务加到 tasks = [] 中

from locust import User, constant

def my_task(user):
    pass

class MyUser(User):
    tasks = [my_task]
    wait_time = constant(1)

3. @task 可选权重参数，用于指定任务的执行比率。

在以下示例中，constant_pacing(0.1) 每0.1s请求一次（也就是每秒10次），其中task_1占 30%，task_2占 70%（也就是每秒请求 task_1 3次，请求 task_2 7次）

from locust import HttpUser, task, constant , constant_pacing

class AdminUser(HttpUser):
    fixed_count = 1
    wait_time = constant_pacing(0.1)
    host = 'http://127.0.0.1:8080'

    @task(30)
    def task_1(self):
        self.client.get("/p/1.html")

    @task(70)
    def task_2(self):
        self.client.get("/p/2.html")

@tag 标签装饰器

@tag 标签装饰器要配合命令行参数 --tags 或 --exclude-tags 一起使用，如下代码

from locust import HttpUser, task, constant , constant_pacing, tag

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @tag('tag1')
    @task
    def task_1(self):
        self.client.get("/p/1.html")

    @tag('tag1', 'tag2')
    @task
    def task_2(self):
        self.client.get("/p/2.html")

    @tag('tag3')
    @task
    def task_3(self):
        self.client.get("/p/3.html")

执行命令 locust -f a1.py --tags tag1 会请求 task_1 和 task_2
执行命令 locust -f a1.py --tags tag2 会请求 task_2
执行命令 locust -f a1.py --exclude-tags tag2 会请求除了 task_2 外的所有请求

on_start 类开始运行前、on_stop 类结束运行前

on_start 在开始运行时将调用其方法
on_stop 在停止运行时将调用其方法

from locust import HttpUser, task

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    def on_start(self):
        print("类似类中的构造方法，每个用户在任务开始前，只执行一次，在这里可以定义一个对象的属性，这样其它测试集就可以使用这个属性")
 
    def on_stop(self):
        print("类似类中的析构方法，每个用户在任务结束前，只执行一次，在这里可以定义一个对象的属性，这样其它测试集就可以使用这个属性")

    @task
    def task_1(self):
        self.client.get("/p/1.html")

test_start 压测开始前、test_stop 压测试结束前

如果需要在压测开始时或停止时运行某些代码，则使用 test_start 和test_stop 事件

from locust import HttpUser, task, events

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @events.test_start.add_listener
    def on_test_start(environment, **kwargs):
        print("测试开始前，执行一次。比如，在测试前需要构造数据等")

    @events.test_stop.add_listener
    def on_test_stop(environment, **kwargs):
        print("测试停止前，执行一次。比如，在测试后需要删除压测数据等")

    @task
    def task_1(self):
        self.client.get("/p/1.html")

init 启动 locust 时触发的

init 事件在每个 Locust 进程开始时触发，只要执行 locust 命令后即运行。这在分布式模式下特别有用。
在分布模式下，如果一个 master 带多个 worker，使用 test_start 只会运行一次，如果这时你想在 worker 机器上执行一些命令就可以使用 init 事件

from locust import HttpUser, task, events
from locust.runners import MasterRunner

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @events.init.add_listener
    def on_locust_init(environment, **kwargs):
        if isinstance(environment.runner, MasterRunner):
            print("我在 master 节点上")
        else:
            print("我在 worker 节点或独立节点上")

    @task
    def task_1(self):
        self.client.get("/p/1.html")

执行命令 locust -f a1.py --master 就会打印 “我在 master 节点上”
执行命令 locust -f a1.py 或 locust -f a1.py --worker 就会打印 “我在 worker 节点或独立节点上”

HttpSession

HttpSession 的主要功能是将请求结果报告给 Locust（success/fail, response time, response length, name）
适用于所有 HTTP 方法的方法：get, post, put, ...

就像 requests.Session 一样，它会在请求之间保留 cookie
例如：登录后自动保存 cookie，这样就可以请求后续需要登录才能访问的页面

from locust import HttpUser, task

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        response = self.client.post("/user/user/login", json={"account": "aaa", "code": "xxx"})
        print("Response status code:", response.status_code)
        print("Response text:", response.text)
        response = self.client.get("/user/user/getInfo")
        print("Response status code:", response.status_code)
        print("Response text:", response.text)

2. 单次修改 sesion

如何登录成功后，在返回 json 中携带 token，后面需要将 token 添加到后续请求的 headr 中，就可以用这个方法
假如请求 /user/user/login 后返回的 json 如下

{
    "code": 0,
    "msg": "请求成功",
    "data": {
        "token": "eyJ0 ...... vH0"
    }
}

则 locust 代码，如下：

其中请求 /user/user/login 后会将返回内容转为 json 对象，然后取得 data / token
将得到的 token 存到 header 中，修改下一个请求 /user/user/getInfo 的 headers，这样就可以正常请求 getInfo 接口了

from locust import HttpUser, task
import json

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        response = self.client.post("/user/user/login", json={"account": "aaa", "code": "xxx", "local": False})
        print("Response status code:", response.status_code)
        print("Response text:", response.text)
        res_token = json.loads(response.text)["data"]["token"]
        print("Response token:", res_token)
        header = {
            "Token" : res_token
        }
        response = self.client.get("/user/user/getInfo", headers=header, verify=False)
        print("Response status code:", response.status_code)
        print("Response text:", response.text)

3. 修改 sesion

上例中在请求 /user/user/getInfo 接口时 headers 中会添加 token，但如果后续还有请求也需要 token 怎么办呢？
如下使用 headers.update 更新 sesion 中的 headers

同样也是先从 /user/user/login 拿到 token，保存到 header 中
使用 self.client.headers.update(header) 更新 HttpSession 中的 headers，这样之后的所有请求 headers 中都会携带 token
使用 print(self.client.headers) 可以打印当前 HttpSession 中的 headers
可以看到调用 update 之前是 {'User-Agent': 'python-requests/2.31.0', 'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Connection': 'keep-alive'} 之后是 {'User-Agent': 'python-requests/2.31.0', 'Accept-Encoding': 'gzip, deflate, br', 'Accept': '*/*', 'Connection': 'keep-alive', 'Token': 'eyJ......Mw'}

from locust import HttpUser, task
import json

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        response = self.client.post("/user/user/login", json={"account": "aaa", "code": "xxx", "local": False})
        print(self.client.headers)
        print("Response status code:", response.status_code)
        print("Response text:", response.text)
        res_token = json.loads(response.text)["data"]["token"]
        print("Response token:", res_token)
        header = {
            "Token" : res_token
        }
        self.client.headers.update(header)
        response = self.client.get("/user/user/getInfo", verify=False)
        print(self.client.headers)
        print("Response status code:", response.status_code)
        print("Response text:", response.text)

Http

get 请求

1. 无参 get 请求

上面已经写过很多 get 示例了，这里就不赘述了，其实就是简单给一个路径

response = self.client.get("/user/user/getInfo", verify=False)

2. 带参 get 请求

可以有两中方式携带 url 参数

直接拼接在路径中

from locust import HttpUser, task

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        response = self.client.get("/user/user/getInfo?uid=aa&name=xx", verify=False)
        print("Response status code:", response.status_code)

使用 params 参数
首先定义参数 query_data，然后通过 params=query_data 设置 url 参数

from locust import HttpUser, task

class AdminUser(HttpUser):
    fixed_count = 1
    host = 'http://127.0.0.1:8080'

    @task
    def task_1(self):
        query_data = {
            'id' : 'aa',
            'name' : 'xx'
        }
        response = self.client.get("/user/user/getInfo", params=query_data, verify=False)
        print("Response status code:", response.status_code)

post 请求

参数

method：新 Request 对象的方法
url：新 Request 对象的URL
name：可选参数。
可以指定为 Locust 的统计信息中的标签，用于代替 URL 路径。这可以用于将被请求的不同 URL 分组到 Locust 统计数据中的一个条目中。
catch_response可选参数。如果要设置，可以是一个布尔值。可以用来使请求返回为作为with 语句的参数的上下文管理器。这将允许根据响应内容将请求标记为失败，即使响应代码是 ok (2xx) ，反之亦然。可以使用 catch_response捕捉请求，然后将其标记为成功，即使响应代码不是 ok (例如 500 或 404)。
params：可选参数。要发送到Request的查询字符串的字典或 bytes 对象。
data：可选参数。要发送到 Request主体中的字典或 bytes 对象。
headers：可选参数。与 Request一起发送的表示 HTTP headers 的字典。
cookies：可选参数。与 Request一起发送的表示 cookies 的 dict 或 CookieJar 对象。
files：可选参数。用于多部分编码上传的元素为 filename: filename: file-like-objects 的字典。
auth：可选参数：用于启用 Basic/Digest或自定义的 HTTP Auth 的元组或可调用对象。
timeout：可选参数。以浮点数或(连接超时、读取超时)元组的形式等待服务器发送数据的时间(以秒为单位)。
allow_redirects：可选参数。布尔类型。默认值为 True。表示是否允许重定向。
proxies：可选参数。字典类型。键表示代理使用的协议，键值表示代理的URL。
stream：可选参数。是否立即下载响应内容。默认值为 False。
verify：可选参数。如果为True，则会验证 SSL 证书。也可以提供一个 CA_BUNDLE 路径。
cert：可选参数。如果提供一个字符串。那么应该是指向SSL 客户端证书（.pem文件）的路径；如果是一个元组，则应该是（‘cert’, ‘key’）。