编写 Locust 文件

现在，让我们来看一个更完整/更现实的例子，展示你的测试可能是什么样的：

import time
from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")
            time.sleep(1)

    def on_start(self):
        self.client.post("/login", json={"username":"foo", "password":"bar"})

让我们逐步分析：

import time
from locust import HttpUser, task, between

Locust 文件只是一个普通的 Python 模块，你可以从其他文件或包中导入代码。

class QuickstartUser(HttpUser):

我们在这里定义了一个类，用来模拟测试中的用户。它继承自 HttpUser，这使得每个用户都可以使用 client 属性，client 是 HttpSession 的一个实例，可以用来向目标系统发送 HTTP 请求。当测试开始时，Locust 会为每个模拟的用户创建一个该类的实例，并在自己的 Greenlet 线程中运行它们。

为了让文件成为一个有效的 locustfile，它必须至少包含一个继承自 User 的类。

wait_time = between(1, 5)

在这个类中，我们定义了一个 wait_time，让模拟的用户在每次任务执行后，等待 1 到 5 秒。有关 wait_time 属性的更多信息，请参阅相关文档。

@task
def hello_world(self):
    self.client.get("/hello")
    self.client.get("/world")

通过 @task 装饰器声明的方法是 Locust 文件的核心。对于每个运行中的用户，Locust 会创建一个 Greenlet（协程或 “微线程”），调用这些方法。任务中的代码是顺序执行的（它就是常规的 Python 代码），所以 /world 不会在 /hello 响应之前被调用。

@task
def hello_world(self):
    ...

@task(3)
def view_items(self):
    ...

我们声明了两个任务，分别用 @task 装饰，其中一个任务的权重为 3。运行时，Locust 会随机选择一个任务执行。在本例中，view_items 被选中的概率是 hello_world 的三倍。每个任务执行完后，用户会等待指定的 wait_time（在本例中是 1 到 5 秒）。然后它会再次选择一个任务执行。

需要注意的是，只有被 @task 装饰的方法会被选中执行，你可以根据需要定义其他内部帮助方法。

self.client.get("/hello")

self.client 属性使得我们可以发送 HTTP 请求，Locust 会记录这些请求。关于如何进行其他类型的请求、验证响应等，可以参考 HttpClient 的相关文档。

HttpUser 不是一个真正的浏览器，因此不会解析 HTML 响应来加载资源或渲染页面。但它会跟踪 cookies。

@task(3)
def view_items(self):
    for item_id in range(10):
        self.client.get(f"/item?id={item_id}", name="/item")
        time.sleep(1)

在 view_items 任务中，我们加载了 10 个不同的 URL，并使用了一个动态的查询参数。为了不让这些请求在 Locust 的统计中显示为 10 个不同的条目，我们通过 name 参数将所有请求归为一类。

def on_start(self):
    self.client.post("/login", json={"username":"foo", "password":"bar"})

此外，我们还声明了一个 on_start 方法。每当一个模拟的用户开始运行时，Locust 会调用这个方法。关于 on_start 和 on_stop 方法的更多信息，请参阅文档。

自动生成 locustfile

你可以使用 har2locust 基于浏览器录制（HAR 文件）自动生成 locustfile。

这对于不习惯编写 locustfile 的初学者非常有用，同时也提供了高度的自定义功能，适用于更高级的用例。

har2locust 仍处于 beta 阶段，可能并不总是生成正确的 locustfile，并且其接口可能会在不同版本之间有所变化。

User 类

一个 User 类代表系统中的一种用户/场景。在进行测试时，你需要指定要模拟的并发用户数，Locust 将为每个用户创建一个实例。你可以在这些类/实例中添加任何你喜欢的属性，但有一些属性对于 Locust 是特别重要的：

`wait_time` 属性

User 的 wait_time 方法使得在每个任务执行之后可以轻松地引入延迟。如果没有指定 wait_time，则下一个任务会在上一个任务完成后立即执行。

constant：固定时间
between：在最小值和最大值之间随机选择的时间

例如，要让每个用户在每个任务执行之间等待 0.5 到 10 秒：

from locust import User, task, between

class MyUser(User):
    @task
    def my_task(self):
        print("executing my_task")

    wait_time = between(0.5, 10)

constant_throughput：一种自适应时间，确保任务每秒执行（最多）X 次。
constant_pacing：一种自适应时间，确保任务每 X 秒执行一次（它是 constant_throughput 的数学反演）。

例如，如果你希望 Locust 在峰值负载时每秒运行 500 次任务迭代，你可以使用 wait_time = constant_throughput(0.1) 和 5000 个用户。

wait_time 只能约束吞吐量，不能启动新用户以达到目标。因此，在我们的示例中，如果任务迭代的时间超过 10 秒，吞吐量将少于 500。

wait_time 在任务执行后应用，因此，如果你有高的启动速率/增加速率，可能会在 ramp-up 期间超过目标。

wait_time 适用于任务，而不是请求。例如，如果你在任务中指定 wait_time = constant_throughput(2) 并且执行了两个请求，则每个用户的请求速率（RPS）将是 4。

你还可以直接在类中声明自定义的 wait_time 方法。例如，以下 User 类将在每次执行后分别等待 1 秒、2 秒、3 秒，依此类推。

class MyUser(User):
    last_wait_time = 0

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

`weight` 和 `fixed_count` 属性

如果 locustfile 中有多个用户类，并且没有在命令行中指定用户类，Locust 将为每个用户类生成相同数量的用户。你也可以通过在命令行中传递它们来指定使用哪些用户类：

locust -f locust_file.py WebUser MobileUser

如果你希望模拟某种类型的用户比另一种更多，可以为这些类设置 weight 属性。下面的代码将使 Locust 生成比 MobileUser 多三倍的 WebUser：

class WebUser(User):
    weight = 3
    ...

class MobileUser(User):
    weight = 1
    ...

你也可以设置 fixed_count 属性。在这种情况下，weight 属性将被忽略，只有确切数量的用户将被生成。这些用户会在任何常规的、加权的用户之前被生成。下面的示例中，只有一个 AdminUser 实例会被生成，用于更精确地控制请求数量，而不依赖于总用户数量。

class AdminUser(User):
    wait_time = constant(600)
    fixed_count = 1

    @task
    def restart_app(self):
        ...

`host` 属性

host 属性是要测试的 URL 前缀（例如 https://google.com）。它会自动添加到请求中，因此你可以像这样使用：self.client.get("/")。

你也可以在 Locust 的 Web UI 或命令行中使用 --host 选项覆盖这个值。

`tasks` 属性

一个 User 类可以通过 @task 装饰器将任务声明为方法，也可以使用 tasks 属性来指定任务，后者会在下面详细描述。

`environment` 属性

environment 属性是对用户正在运行的环境的引用。你可以用它与环境或其中的运行器进行交互。例如，在任务方法中停止运行器：

self.environment.runner.quit()

如果在独立的 locust 实例上运行，这将停止整个运行。如果在工作节点上运行，它将停止该节点。

`on_start` 和 `on_stop` 方法

用户（和 TaskSets）可以声明 on_start 方法和/或 on_stop 方法。当一个用户开始运行时，它会调用 on_start 方法，当它停止运行时，会调用 on_stop 方法。对于 TaskSet，当模拟用户开始执行该 TaskSet 时，会调用 on_start 方法；当模拟用户停止执行该 TaskSet 时（调用 interrupt() 或用户被杀死时），会调用 on_stop 方法。

Tasks

当负载测试启动时，Locust 会为每个模拟的用户创建一个 User 类的实例，并且它们会在自己的 greenlet 中开始运行。这些用户运行时会选择执行任务、休眠一段时间，然后选择一个新任务，依此类推。

`@task` 装饰器

为 User 添加任务的最简单方法是使用 @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)

@task 装饰器接受一个可选的 weight 参数，用来指定任务的执行比例。在以下示例中，task2 被选择的概率是 task1 的两倍：

from locust import User, task, between

class MyUser(User):
    wait_time = between(5, 15)

    @task(3)
    def task1(self):
        pass

    @task(6)
    def task2(self):
        pass

`tasks` 属性

另一种定义 User 任务的方法是设置 tasks 属性。

tasks 属性可以是一个任务列表，或者是一个 <Task : int> 字典，其中 Task 是一个 Python 可调用对象或一个 TaskSet 类。如果任务是一个普通的 Python 函数，它们接收一个参数，即正在执行该任务的 User 实例。

这是一个将任务声明为普通 Python 函数的示例：

from locust import User, constant

def my_task(user):
    pass

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

如果 tasks 属性被指定为一个列表，那么每次执行任务时，它都会从 tasks 属性中随机选择一个任务。如果 tasks 是一个字典（键为可调用任务，值为整数），则会随机选择一个任务，且选择的概率由整数值的大小决定。例如，像这样定义的任务：

{my_task: 3, another_task: 1}

在这种情况下，my_task 被选择的概率是 another_task 的三倍。

实际上，上述字典会被扩展成一个列表（tasks 属性被更新），如下所示：

[my_task, my_task, my_task, another_task]

然后，使用 Python 的 random.choice() 从这个列表中选择任务。

`@tag` 装饰器

通过使用 @tag 装饰器为任务打标签，你可以在测试过程中有选择地执行某些任务，使用 --tags 和 --exclude-tags 参数。考虑以下示例：

from locust import User, constant, task, tag

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

    @tag('tag1')
    @task
    def task1(self):
        pass

    @tag('tag1', 'tag2')
    @task
    def task2(self):
        pass

    @tag('tag3')
    @task
    def task3(self):
        pass

    @task
    def task4(self):
        pass

如果你使用 --tags tag1 启动这个测试，那么只有 task1 和 task2 会被执行。如果你使用 --tags tag2 tag3 启动测试，那么只有 task2 和 task3 会被执行。

--exclude-tags 的行为则完全相反。所以，如果你使用 --exclude-tags tag3 启动测试，那么只有 task1、task2 和 task4 会被执行。排除标签优先于包含标签，因此，如果一个任务有你已经包含的标签，并且有一个你排除的标签，这个任务将不会被执行。

Events

如果你想在测试中运行一些初始化代码，通常只需将其放在 locustfile 的模块级别，但有时你需要在特定的时刻执行某些操作。为此，Locust 提供了事件钩子。

`test_start` 和 `test_stop`

如果你需要在负载测试的开始或结束时运行一些代码，可以使用 test_start 和 test_stop 事件。你可以在 locustfile 的模块级别设置这些事件的监听器：

from locust import events

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("A new test is starting")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("A new test is ending")

`init` 事件

init 事件会在每个 Locust 进程开始时触发。这对于分布式模式尤其有用，因为在每个用户启动之前，每个工作进程（而不是每个用户）需要有机会执行初始化。例如，假设你有一些全局状态，所有从该进程启动的用户都需要使用：

from locust import events
from locust.runners import MasterRunner

@events.init.add_listener
def on_locust_init(environment, **kwargs):
    if isinstance(environment.runner, MasterRunner):
        print("I'm on master node")
    else:
        print("I'm on a worker or standalone node")

其他事件

有关其他事件和如何使用它们的更多示例，请参见通过事件钩子扩展 Locust。

HttpUser class

HttpUser 是最常用的用户类。它添加了一个 client 属性，用于发送 HTTP 请求。

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(5, 15)

    @task(4)
    def index(self):
        self.client.get("/")

    @task(1)
    def about(self):
        self.client.get("/about/")

`client` 属性 / `HttpSession`

client 是 HttpSession 的一个实例。HttpSession 是 requests.Session 的子类/封装类，因此其功能文档丰富，许多人应该很熟悉。HttpSession 主要添加的是将请求结果（如成功/失败、响应时间、响应长度、名称等）报告给 Locust。

它包含所有 HTTP 方法的实现：如 get、post、put 等。

就像 requests.Session 一样，HttpSession 会在请求之间保留 cookies，因此可以轻松用于网站登录。

例如，发送一个 POST 请求，查看响应，并隐式地重用我们为第二个请求获得的会话 cookie：

response = self.client.post("/login", {"username":"testuser", "password":"secret"})
print("Response status code:", response.status_code)
print("Response text:", response.text)
response = self.client.get("/my-profile")

HttpSession 会捕获任何由 Session 引发的 requests.RequestException（由连接错误、超时等引起），而返回一个虚拟的 Response 对象，status_code 设置为 0，content 设置为 None。

验证响应

默认情况下，如果 HTTP 响应码小于 400，则认为请求成功，但通常需要做一些额外的响应验证。

你可以使用 catch_response 参数、with 语句和 response.failure() 来标记请求失败：

with self.client.get("/", catch_response=True) as response:
    if response.text != "Success":
        response.failure("Got wrong response")
    elif response.elapsed.total_seconds() > 0.5:
        response.failure("Request took too long")

即使响应码不好，你也可以通过以下方式将请求标记为成功：

with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

你甚至可以通过抛出异常来完全避免记录某个请求，之后在 with 语句外部捕获它，或者可以抛出 Locust 异常，让 Locust 捕获它：

from locust.exception import RescheduleTask
...
with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        raise RescheduleTask()

REST/JSON APIs

FastHttpUser 提供了现成的 rest 方法，但你也可以手动实现：

from json import JSONDecodeError
...
with self.client.post("/", json={"foo": 42, "bar": None}, catch_response=True) as response:
    try:
        if response.json()["greeting"] != "hello":
            response.failure("Did not get expected value in greeting")
    except JSONDecodeError:
        response.failure("Response could not be decoded as JSON")
    except KeyError:
        response.failure("Response did not contain expected key 'greeting'")

分组请求

网站上常见的情况是，页面的 URL 中包含某种动态参数。在这种情况下，将这些 URL 分组在一起进行统计很有意义。可以通过向 HttpSession 的不同请求方法传递 name 参数来实现：

# 这些请求的统计数据将被分组为：/blog/?id=[id]
for i in range(10):
    self.client.get("/blog?id=%i" % i, name="/blog?id=[id]")

有些情况下，无法将参数传递给请求函数（例如，使用封装 requests 会话的库/SDK）。在这种情况下，可以通过设置 client.request_name 属性来进行分组：

# 这些请求的统计数据将被分组为：/blog/?id=[id]
self.client.request_name = "/blog?id=[id]"
for i in range(10):
    self.client.get("/blog?id=%i" % i)
self.client.request_name = None

如果你希望通过最小的样板代码进行多个分组，可以使用 client.rename_request() 上下文管理器：

@task
def multiple_groupings_example(self):
    # 这些请求的统计数据将被分组为：/blog/?id=[id]
    with self.client.rename_request("/blog?id=[id]"):
        for i in range(10):
            self.client.get("/blog?id=%i" % i)

    # 这些请求的统计数据将被分组为：/article/?id=[id]
    with self.client.rename_request("/article?id=[id]"):
        for i in range(10):
            self.client.get("/article?id=%i" % i)

使用 catch_response 和直接访问 request_meta，你甚至可以基于响应中的某些内容重命名请求：

with self.client.get("/", catch_response=True) as resp:
    resp.request_meta["name"] = resp.json()["name"]

HTTP 代理设置

为了提高性能，我们配置 requests 不再查找环境中的 HTTP 代理设置，将 requests.Session 的 trust_env 属性设置为 False。如果你不希望这样做，可以手动将 locust_instance.client.trust_env 设置为 True。有关详细信息，请参见 requests 的文档。

连接重用

默认情况下，HttpUser 会重用连接，甚至跨任务运行。如果你想避免连接重用，可以这样做：

self.client.get("/", headers={"Connection": "close"})
self.client.get("/new_connection_here")

或者，你也可以关闭整个 requests.Session 对象（这会删除 cookies、关闭 SSL 会话等）。这样会有一些 CPU 开销（下一个请求的响应时间会更高，因为需要重新协商 SSL 等），因此除非真的需要，否则不要使用这种方法：

self.client.get("/")
self.client.close()
self.client.get("/new_connection_here")

连接池

每个 HttpUser 都会创建新的 HttpSession，因此每个用户实例都有自己的连接池。这与真实用户（浏览器）与 Web 服务器的交互方式类似。

如果你希望共享连接池，可以使用一个单独的池管理器。为此，可以将 pool_manager 类属性设置为 urllib3.PoolManager 的实例：

from locust import HttpUser
from urllib3 import PoolManager

class MyUser(HttpUser):
    # 所有该类的实例最多会限制为 10 个并发连接。
    pool_manager = PoolManager(maxsize=10, block=True)

有关更多配置选项，请参阅 urllib3 的文档。

TaskSets

TaskSets 是一种结构化测试层次化网站/系统的方法。你可以在这里了解更多内容。

示例

这里有很多 Locust 文件的示例。

如何构建你的测试代码

重要的是要记住，locustfile.py 只是一个普通的 Python 模块，由 Locust 导入。在这个模块中，你可以像在任何 Python 程序中一样自由地导入其他 Python 代码。当前工作目录会自动添加到 Python 的 sys.path 中，因此工作目录中的任何 Python 文件/模块/包都可以使用 Python 的 import 语句导入。

对于小型测试，将所有测试代码保存在单个 locustfile.py 文件中应该可以正常工作，但对于较大的测试套件，你可能希望将代码拆分为多个文件和目录。

你如何构建测试源代码的结构完全取决于你，但我们建议你遵循 Python 的最佳实践。以下是一个虚构的 Locust 项目文件结构示例：

Project root
- common/
  - __init__.py
  - auth.py
  - config.py
- locustfile.py
- requirements.txt # 外部 Python 依赖项通常保存在 requirements.txt 中

如果项目有多个 locustfile，可以将它们保存在一个单独的子目录中：

Project root
- common/
  - __init__.py
  - auth.py
  - config.py
- my_locustfiles/
  - api.py
  - website.py
- requirements.txt

在以上任何一种项目结构中，你的 locustfile 都可以通过以下方式导入公共库：

import common.auth

编写 Locust 文件

自动生成 locustfile

User 类

wait_time 属性

weight 和 fixed_count 属性

host 属性

tasks 属性

environment 属性

on_start 和 on_stop 方法

Tasks

@task 装饰器

tasks 属性

@tag 装饰器

Events

test_start 和 test_stop

init 事件

其他事件

HttpUser class

client 属性 / HttpSession

验证响应

REST/JSON APIs

分组请求

HTTP 代理设置

连接重用

连接池

TaskSets

示例

如何构建你的测试代码

`wait_time` 属性

`weight` 和 `fixed_count` 属性

`host` 属性

`tasks` 属性

`environment` 属性

`on_start` 和 `on_stop` 方法

`@task` 装饰器

`tasks` 属性

`@tag` 装饰器

`test_start` 和 `test_stop`

`init` 事件

`client` 属性 / `HttpSession`