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，HttpUser为每个用户提供了一个client属性，该属性是HttpSession的一个实例，可用于向我们要负载测试的目标系统发出HTTP请求。当测试开始时，locust将为它模拟的每个用户创建这个类的一个实例，每个用户都将开始在他们自己的gevent线程中运行。 要使文件成为有效的 locustfile，它必须至少包含一个继承自 User 的类。

wait_time = between(1, 5)

我们的类定义了一个 wait_time ，这将使模拟用户在每个任务（见下文）执行后等待 1 到 5 秒。更多详细信息，请参阅 wait_time 属性。

@task
def hello_world(self):
    ...

使用用 @task 修饰的方法是 locust 文件的核心。对于每个正在运行的用户，Locust 都会创建一个 greenlet（微线程）去调用这些方法。

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

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

我们通过用@task修饰两个方法声明了两个任务，其中一个被赋予了更高的权重：3。 当我们的QuickstartUser运行时，它将选择一个声明的任务并执行它-在本例中是hello_world或view_items。任务是随机挑选的，但你可以给它们不同的权重。上述配置将使Locust选择view_item的可能性是hello_world的三倍。当任务执行完毕后，用户将在其等待时间(在本例中为1到5秒)期间休眠。等待时间过后，它会选择一项新任务，并不断重复该任务。

请注意，只会选择使用 @task 修饰的方法，因此您可以以任何您喜欢的方式定义自己的内部辅助方法。

self.client.get("/hello")

self.client 属性使得Locust 请求HTTP 成为可能。有关如何发出其他类型的请求、验证响应等的信息，请参阅使用 HTTP 客户端。

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

@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 个单独的条目（因为统计信息是在 URL 上分组的），我们使用 name 参数将所有这些请求分组到一个名为“/item”的条目下。

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

此外，我们还声明了一个 on_start 方法。当每个模拟用户启动时，将调用具有此名称的方法。有关详细信息，请参阅 on_start 和 on_stop 方法。

User class

一个User 类代表一个用户（如果你愿意，也可以是一群蝗虫）。 Locust 将为正在模拟的每个用户生成一个 User 类的实例。User 类可以定义一些通用属性。

wait_time attribute

用户的 wait_time 方法在每次任务执行后引入延迟。如果未指定 wait_time，则下一个任务将在完成后立即执行。

• constant： 在固定的时间内
• between：在最小值和最大值之间的随机时间
• constant_throughput：用于确保任务每秒运行（最多）X 次的自适应时间。
• constant_pacing：用于确保任务每 X 秒（最多）运行一次的自适应时间（它是 constant_throughput 的数学倒数）

例如，让每个用户在每次任务执行之间等待 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)

• 例如，如果您希望 Locust 在峰值负载下每秒运行 500 次任务迭代，您可以设置 wait_time = constant_throughput(0.1) 和 用户数为5000。
• wait_time只能限制吞吐量，不能启动新用户以达到目标。因此，在我们的示例中，如果任务迭代的时间超过 10 秒，吞吐量将小于 500。
• wait_time在任务执行后应用，因此如果您有一个很高的生成速率，您最终可能会在加速期间超过您的目标。
• wait_time适用于task，而不是reqeust。例如，如果您指定 wait_time = constant_throughput(2) 并在您的任务中执行两个请求，每个用户的请求率/RPS 将为 4。

可以直接在你的类上声明你自己的 wait_time 方法。例如，下面的 User 类会休眠一秒钟，然后是两秒钟，然后是三秒钟，以此类推。

class MyUser(User):
    last_wait_time = 0

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

    ...

weight and fixed_count attributes

如果文件中存在多个User 类，并且在命令行上没有指定 User 类，Locust 将为每个User类生成相同数量的用户。您还可以通过将它们作为命令行参数传递来指定要使用同一 locustfile 中的哪些User类：

locust -f locust_file.py WebUser MobileUser

如果您希望模拟更多特定类型的用户，您可以在这些类上设置权重属性。例如，网络用户的可能性是移动用户的三倍：

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

您还可以设置 fixed_count 属性。在这种情况下，权重属性将被忽略，并且将产生精确数量的用户。这些用户首先产生。在下面的示例中，将生成 AdminUser 的唯一实例以进行一些特定的工作，更准确地控制请求次数，而与总用户数无关。

?

测试结果：如果总用户不足100，那么只有AdminUser才会工作，而WebUser无法工作。

?

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

    @task
    def restart_app(self):
        ...

class WebUser(User):
    ...

host attribute

host属性是要加载的主机的 URL 前缀（即“http://www.boxiaoyi.com”）。通常，这是在 Locust 的 Web UI 或命令行中指定的，在 locust 启动时使用--host选项。 如果在User 类中声明了一个host属性，它将在命令行或网络请求中没有指定 --host 的情况下使用。

class WebUser(User):
    host='http://www.boxiaoyi.com'

tasks attribute

User 类可以使用 @task 装饰器将任务声明为方法，但也可以使用 tasks 属性指定任务，下面将详细介绍。

environment attribute

对用户运行environment的引用。使用它与环境或其包含的runner进行交互。例如:从任务方法中停止runner（运行器）：

self.environment.runner.quit()

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

on_start and on_stop methods

Users（和TaskSets）可以声明 on_start 方法和 on_stop 方法。

• 对于User 类，将在开始运行时调用其 on_start 方法，并在停止运行时调用其 on_stop 方法。
• 对于TaskSet， on_start 方法在模拟用户开始执行该TaskSet 时调用， on_stop 方法在模拟用户停止执行该TaskSet 时调用（当调用 interrupt() 或用户被杀死时）。

Tasks

当负载测试开始时，将为每个模拟用户创建一个 User 类的实例，并且他们将开始在自己的绿色线程中运行。当这些用户运行时，它们执行选择的任务，休眠一段时间，然后选择一个新任务，依此类推。 这些任务是普通的 Python 可调用对象，并且 - 如果我们正在对拍卖网站进行负载测试 - 它们可以执行诸如“加载起始页”、“搜索某些产品”、“出价”等操作。

@task decorator

为用户添加任务的最简单方法是使用@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 接受一个可选的权重参数，可用于指定任务的执行率。在以下示例中，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 attribute

定义用户任务的另一种方法是设置task属性。 tasks 属性可以是一个 Task 列表，也可以是一个 dict（格式： ），其中 Task 可以是 python 可调用对象，也可以是 TaskSet 类。如果任务是普通的 python 函数，他们会收到一个参数，即正在执行任务的用户实例。

这是一个声明为普通 python 函数的用户任务的示例：

from locust import User, constant

def my_task(user):
    pass

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

• 如果将任务属性指定为列表，则每次执行任务时，将从任务属性中随机选择。
• 但是，如果任务是一个字典——以可调用对象作为键，以整数作为值——将随机选择要执行的任务，但以整数作为权重。所以有一个看起来像这样的任务：

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


    def task1(self):
        pass


    def task2(self):
        pass
    {task1: 3, task2: 1}

task1 执行的可能性是 task2 的 3 倍。

在内部，上面的 dict 实际上将扩展为一个列表（并且更新了 tasks 属性），如下所示：

[my_task, my_task, my_task, another_task]

然后使用 Python 的 random.choice() 从列表中挑选任务。

@tag decorator

通过使用 @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。--exclude-tags优先级高于--tags，所以如果一个任务有一个你已经包含的标签和一个你已经排除的标签，它将不会被执行。

Events

如果您想在测试中运行一些设置代码，通常将其放在 locustfile 的模块级别就足够了，但有时您需要在运行中的特定时间执行某些操作。为此，Locust 提供了事件挂钩。

test_start and 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 进程开始时触发。这在分布式模式中特别有用，在这种模式下，每个worker进程（而不是每个用户）都需要有机会进行一些初始化。例如，假设你有一些全局状态，所有从这个进程产生的用户都需要：

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")

Other events

请参阅使用事件挂钩扩展 locust 以了解其他事件以及如何使用它们的更多示例。

HttpUser 类

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 attribute / HttpSession

client是 HttpSession 的一个实例。 HttpSession 是 requests.Session 的子类/包装器，因此它的功能有据可查，许多人应该很熟悉。 HttpSession 主要是将请求结果报告给 Locust（成功/失败、响应时间、响应长度、名称）。

它包含所有 HTTP 方法的方法：get、post、put……

就像 requests.Session 一样，它在请求之间保留 cookie，因此可以很容易地用于登录网站。

发出 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（由连接错误、超时或类似原因引起），而不是返回一个 status_code 设置为 0 且内容设置为 None 的虚拟 Response 对象。

Validating responses

如果 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")
    elif 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

下面是一个如何调用 REST API 并验证响应的示例：

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'")

locust-plugins 有一个现成的用于测试 REST API 的类，称为 RestUser

Grouping requests

网站的 URL 包含某种动态参数的页面很常见。通常在用户统计中将这些 URL 组合在一起是有意义的。这可以通过将名称参数传递给 HttpSession 的不同请求方法来完成。

# Statistics for these requests will be grouped under: /blog/?id=[id]
for i in range(10):
    self.client.get("/blog?id=%i" % i, name="/blog?id=[id]")

在某些情况下，无法将参数传递给请求函数，例如与包装请求会话的库/SDK 交互时。另外一种做法是通过设置client.request_name属性提供了分组请求：

# Statistics for these requests will be grouped under: /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):

    # Statistics for these requests will be grouped under: /blog/?id=[id]
    with self.client.rename_request("/blog?id=[id]"):
        for i in range(10):
            self.client.get("/blog?id=%i" % i)

    # Statistics for these requests will be grouped under: /article/?id=[id]
    with self.client.rename_request("/article?id=[id]"):
        for i in range(10):
            self.client.get("/article?id=%i" % i)

HTTP Proxy settings

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

TaskSets（任务集）

TaskSets 是一种构建分层网站/系统测试的方法。你可以在这里读更多关于它的内容.

如果以分层方式构建的网站进行性能测试，其中包含部分和子部分，那么以相同的方式构建负载测试可能会很有用。

TaskSets 是一项高级功能，很少有用。大多数时候，最好使用常规的 Python 循环和控制语句来实现相同的目标。还有一些陷阱，最常见的是忘记调用 self.interrupt().

from locust import User, TaskSet, constant

class ForumSection(TaskSet):
    wait_time = constant(1)

    @task(10)
    def view_thread(self):
        pass

    @task
    def create_thread(self):
        pass

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

class LoggedInUser(User):
    wait_time = constant(5)
    tasks = {ForumSection:2}

    @task
    def my_task(self):
        pass

TaskSet 也可以使用@task 装饰器直接内联在 User/TaskSet 类下：

class MyUser(User):
    @task
    class MyTaskSet(TaskSet):
        ...

TaskSet 类的任务可以是其他 TaskSet 类，允许它们嵌套任意数量的级别。这使我们能够以更真实的方式定义模拟用户的行为。

例如，我们可以定义具有以下结构的任务集：

- Main user behaviour
  - Index page
  - Forum page
    - Read thread
      - Reply
    - New thread
    - View next page
  - Browse categories
    - Watch movie
    - Filter movies
  - About page

当正在运行的用户线程选择要执行的任务集类时，将创建此类的一个实例，然后执行将进入此任务集。然后发生的事情是 TaskSet 的任务之一将被选中并执行，然后线程将休眠用户的 wait_time 函数指定的持续时间（除非wait_time函数已直接在 TaskSet 类上声明，在这种情况下它将请改用该函数），然后从 TaskSet 的任务中选择一个新任务，再次等待，依此类推。

taskset实例包含对用户的引用- self.user。 它还具有其用户的客户端属性的快捷方式。 因此，您可以使用 self.client.request() 进行请求，就像您的任务直接在HttpUser上定义。

中断任务集

关于 TaskSet 需要了解的一件重要的事情是，它们永远不会停止执行其任务，并且会自行将执行交还给其父 User/TaskSet。这必须由开发人员通过调用该TaskSet.interrupt()方法来完成。 「interrupt(」「self」「, 「「reschedule=True」」)」

• 中断 TaskSet 并将执行控制权交还给父 TaskSet。
• 如果 reschedule 为 True（默认），则父用户将立即重新安排并执行新任务。

在下面的示例中，如果我们没有调用 的停止任务self.interrupt()，则模拟用户一旦进入论坛任务集就永远不会停止运行论坛任务集中的任务：

class RegisteredUser(User):
    @task
    class Forum(TaskSet):
        @task(5)
        def view_thread(self):
            pass

        @task(1)
        def stop(self):
            self.interrupt()

    @task
    def frontpage(self):
        pass

使用中断函数，我们可以与任务加权一起定义模拟用户离开论坛的可能性。

TaskSet 和 User 类中tasks的区别

与直接在User 类下的任务相比，驻留在Taskset下的任务有所差异。它们在执行时通过的参数（作为@Task装饰器的方法声明为方法）是对Taskset实例的引用 ，而不是User 实例。 User 实例可以通过Taskset.User从Taskset实例中访问。 Tasksets还包含一个方便client属性，指的是 User 实例上的 client 属性。

引用 User 实例或父 TaskSet 实例

TaskSet 实例的属性 user 指向它的 User 实例，属性 parent 指向它的父 TaskSet 实例。

Tags and TaskSets

您可以使用@tag 装饰器标记任务集，与标记task类似，但有一些细微差别。 标记 TaskSet 将自动将标记应用于 TaskSet 的所有任务。此外，如果您在嵌套的 TaskSet 中标记任务，即使未标记 TaskSet，Locust 也会执行该任务。

SequentialTaskSet class

SequentialTaskSet 是一个 TaskSet，其任务将按照声明的顺序执行。可以将 SequentialTaskSets 嵌套在 TaskSet 中，反之亦然。 比如下面的代码会按顺序请求URLs /1-/4，然后重复。

def function_task(taskset):
    taskset.client.get("/3")

class SequenceOfTasks(SequentialTaskSet):
    @task
    def first_task(self):
        self.client.get("/1")
        self.client.get("/2")

    # you can still use the tasks property to specify a list of tasks
    tasks = [function_task]

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

请注意，您不需要 SequentialTaskSets 来按顺序执行一些请求。在单个任务中完成整个用户流程通常更容易。

How to structure your test code

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

对于小型测试，将所有测试代码保存在单个 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