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，它为每个用户提供一个属性，该属性是HttpSession的一个实例，可用于向我们要进行负载测试的目标系统发出HTTP请求。当测试开始时，locust 将为它模拟的每个用户创建一个此类的实例，并且每个用户都将开始在自己的绿色 gevent 线程中运行。client

要使文件成为有效的 locustfile，它必须包含至少一个从 User 继承的类。

wait_time = between(1, 5)
我们的类定义了一个，该命令将使模拟用户在每个任务（见下文）执行后等待 1 到 5 秒。有关详细信息，请参阅wait_time属性。wait_time

@task
def hello_world(self):
…
装饰是蝗虫文件的核心。对于每个正在运行的用户，Locust 都会创建一个 greenlet（微线程），该线程将调用这些方法。@task

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

@task(3)
def view_items(self):
…
我们通过使用 来修饰两个方法来声明两个任务，其中一个方法被赋予了更高的权重 （3）。当我们运行时，它将选择一个声明的任务 - 在本例中为或 - 并执行它。任务是随机选择的，但您可以为它们提供不同的权重。上述配置将使 Locust 的拾取概率是 的三倍。当任务完成执行后，用户将在其等待时间内休眠（在本例中为 1 到 5 秒）。等待时间过后，它会选择一个新任务并继续重复。@taskQuickstartUserhello_worldview_itemsview_itemshello_world

请注意，将仅选取修饰的方法，因此您可以按照自己喜欢的任何方式定义自己的内部帮助程序方法。@task

self.client.get("/hello")
该属性使得可以进行将由Locust记录的HTTP调用。有关如何发出其他类型的请求、验证响应等的信息，请参阅使用 HTTP 客户端。self.client

注意

HttpUser不是一个真正的浏览器，因此不会解析HTML响应来加载资源或呈现页面。不过，它会跟踪饼干。

@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)
在任务中，我们使用变量查询参数加载 10 个不同的 URL。为了不在Locust的统计信息中获得10个单独的条目 - 因为统计信息是在URL上分组的 - 我们使用name参数将所有这些请求分组到一个名为条目下。view_items"/item"

def on_start(self):
self.client.post("/login", json={“username”:“foo”, “password”:“bar”})
此外，我们还声明了一个on_start方法。当每个模拟用户启动时，将调用具有此名称的方法。有关详细信息，请参阅on_start和on_stop方法。

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

wait_time属性
用户wait_time方法使得在每次任务执行后很容易引入延迟。如果未指定wait_time，则一旦完成一个任务，将立即执行下一个任务。

在固定时间内保持恒定

介于最小值和最大值之间的随机时间之间

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

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

等待时间在任务执行后应用，因此，如果您的生成率/斜坡上升率较高，则最终可能会在升级期间超过目标。

等待时间适用于任务，而不是请求。例如，如果您指定 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

...

重量和fixed_count属性
如果文件中存在多个用户类，并且在命令行上未指定任何用户类，则 Locust 将生成每个用户类的相等数量。您还可以通过将用户类作为命令行参数传递来指定要从同一 locustfile 使用的用户类：

locust -f locust_file.py WebUser MobileUser
如果您希望模拟更多特定类型的用户，则可以在这些类上设置权重属性。例如，Web 用户的可能性是移动用户的三倍：

class WebUser(User):
weight = 3
…

class MobileUser(User):
weight = 1
…
您还可以设置fixed_count属性。在这种情况下，权重属性将被忽略，并且将生成确切的用户计数。这些用户首先生成。在下面的示例中，将生成 AdminUser 的唯一实例，以进行一些特定的工作，并独立于总用户计数更准确地控制请求计数。

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

@task
def restart_app(self):
    ...

class WebUser(User):
…
主机属性
host 属性是要加载的主机的 URL 前缀（即"http://google.com"）。通常，这是在 Locust 的 Web UI 中或在命令行上指定的，使用该选项，当 Locust 启动时。–host

如果在用户类中声明了 host 属性，则在命令行或 Web 请求中指定 no 的情况下将使用它。–host

任务属性
User 类可以使用@task装饰器将任务声明为其下的方法，但也可以使用 tasks 属性指定任务，这将在下面的更多详细信息中介绍。

环境属性
对运行用户的环境的引用。使用它来与环境或其包含的运行器进行交互。例如，从任务方法停止运行者：

self.environment.runner.quit()
如果在独立 locust 实例上运行，这将停止整个运行。如果在工作线程节点上运行，它将停止该特定节点。

on_start和on_stop方法
用户（和 TaskSet）可以声明on_start方法和/或on_stop方法。用户在开始运行时将调用其on_start方法，在停止运行时调用其on_stop方法。对于 TaskSet，当模拟用户开始执行该 TaskSet 时调用 on_start 方法，当模拟用户停止执行该 TaskSet 时（当调用 interrupt（） 或杀死用户时，将调用on_stop。

任务
启动负载测试时，将为每个模拟用户创建一个 User 类的实例，并且这些用户将开始在自己的绿色线程中运行。当这些用户运行时，他们选择他们执行的任务，睡眠一段时间，然后选择一个新任务等等。

这些任务是普通的python可调用对象，如果我们对拍卖网站进行负载测试，他们可以做一些事情，比如"加载起始页"，“搜索某些产品”，"出价"等。

@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采用可选的权重参数，可用于指定任务的执行比率。在以下示例中，任务 2 被选为任务 1 的几率将是其两倍：

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 属性。

tasks 属性可以是 Tasks 的列表，也可以是 <Task ： int> 字典，其中 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)
如果将 tasks 属性指定为列表，则每次执行任务时，都会从 tasks 属性中随机选择该属性。但是，如果任务是一个字典 - 可调用对象作为键，整数作为值 - 则要执行的任务将被随机选择，但int作为比率。因此，对于如下所示的任务：

{my_task: 3, another_task: 1}
my_task被执行的可能性是another_task的3倍。

在内部，上述字典实际上将扩展为一个列表（并且属性已更新），如下所示：tasks

[my_task, my_task, my_task, another_task]
然后使用Python从列表中选择任务。random.choice()

@tag装饰师
通过使用@tag装饰器标记任务，您可以挑剔在测试期间使用 and 参数执行的任务。请考虑以下示例：–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

如果使用 开始此测试，则在测试期间将仅执行 task1 和 task2。如果以 开始，则只会执行 task2 和 task3。–tags tag1–tags tag2 tag3

–exclude-tags将以完全相反的方式表现。因此，如果使用 开始测试，则只会执行 task1、task2 和 task4。排除始终胜过包含，因此，如果任务具有已包含的标记和已排除的标记，则不会执行该任务。–exclude-tags tag3

事件
如果要在测试中运行一些安装代码，通常只需将其放在 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”)
初始化
该事件在每个 Locust 进程开始时触发。这在分布式模式下特别有用，其中每个工作进程（而不是每个用户）都需要有机会执行一些初始化。例如，假设您有一些全局状态，从此进程生成的所有用户都需要该状态：init

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是最常用的用户。它添加了一个用于发出 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/")

客户端属性 / HttpSession
client 是 HttpSession 的一个实例。HttpSession是请求的子类/包装器。会话，因此其功能已得到充分记录，并且应该为许多人所熟悉。HttpSession 添加的内容主要是将请求结果报告给 Locust（成功/失败、响应时间、响应长度、名称）。

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

就像请求一样。会话，它在请求之间保留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 捕获任何请求。会话引发的请求异常（由连接错误、超时或类似情况引起），而是返回一个虚拟响应对象，其中status_code设置为 0，内容设置为 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 块外部捕获它来避免记录请求。或者，您可以抛出一个蝗虫异常，如下面的示例所示，并让蝗虫捕获它。

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 API
下面是如何调用 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：s，称为RestUser。

对请求进行分组
对于网站来说，其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 代理设置
为了提高性能，我们通过设置请求将请求配置为不在环境中查找 HTTP 代理设置。会话trust_env属性为 。如果您不希望这样做，可以手动设置为 。有关更多详细信息，请参阅请求文档。Falselocust_instance.client.trust_envTrue

任务集
TaskSets 是一种构建分层网站/系统测试的方法。您可以在此处阅读更多相关信息

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

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

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

项目根目录

common/

init.py

auth.py

config.py

locustfile.py

requirements.txt（外部Python依赖项通常保存在需求.txt）

具有多个不同蝗虫文件的项目也可以将它们保存在单独的子目录中：

项目根目录

common/

init.py

auth.py

config.py

my_locustfiles/

api.py

website.py

requirements.txt

使用上述任何项目结构，您的 locustfile 可以使用以下命令导入公共库：