一、安装

1、旧版本windows安装：pip3 install locustio

报错：Locust package has moved from 'locustio' to 'locust'. Please update your reference (or pin your version to 0.14.6 if you dont want to update to 1.0)

解决方法：安装最新（发布前）即预发布的locust

pip3 install -U setuptools
​
pip3 install -U --pre locustio

备注：上述安装步骤的是'0.n'版本，而 locust 在2020locust 进入了 '1.n' 版本时代。

2、新版本安装命令统一为：pip3 install locust

备注：如果需要最新最好的 Locust 版本并且不能等待下一个正式的版本时，可安装预发布版本

pip3 install -U --pre locust

二、入门

1、一个简单的例子

该用户将向 /hello，/world发出请求

from locust import HttpUser, task
​
class HelloWorldUser(HttpUser):
    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

将代码放在当前目录中名为locustfile.pylocust的文件中并运行

D:\auto\locus>locust
[2022-05-11 11:13:43,010] admin-PC/INFO/locust.main: Starting web interface at http://0.0.0.0:8089 (accepting connections from all network interfaces)
[2022-05-11 11:13:43,047] admin-PC/INFO/locust.main: Starting Locust 2.8.6

启动 Locust 后，访问http://localhost:8089

字段说明：

Number of users (peak concurrency)：用户峰值，即模拟用户总数

Spawn rate (users started/second)：每秒启动用户数

Host (e.g. http://www.baidu.com)：被测试服务

启动测试后，针对 20 个并发用户，加速速度为 1 个用户/秒，将其指向响应/hello和的服务器/world。

Locust 还可以将结果可视化为图表，显示每秒请求数 (RPS) 等内容：

响应时间（以毫秒为单位）：

用户数量：

2、一个完整的测试示例

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

2.1、 用户类

一个用户类代表一个用户，Locust 将为正在模拟的每个用户生成一个 User 类的实例

2.2、 wait_time属性

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

• constant在固定的时间内
• between最小值和最大值之间的随机时间
• constant_throughput用于确保任务每秒运行（最多）X 次的
• constant_pacing用于确保任务每 X 秒（最多）运行一次（它等于1/constant_throughput）

如果需要限定locust 在峰值负载下每秒运行 5 次任务迭代，可以使用wait_time = constant_throughput(0.1)，表示每秒运行（最多）0.1次的自适应时间，用户数设定为50，Spawn rate=10，效果如下图

等待时间只能限制吞吐量，不能启动新用户以达到目标，因此在上面的示例中，待用户数全部启动后，吞吐率将小于5

等待时间适用于任务，而不是请求。例如，如果指定wait_time = constant_throughput(2)并在的务中执行两个请求，请求率/RPS 将为 4。

自定义wait_time

class MyUser(User):
    last_wait_time = 0
​
    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

2.3、weight 和 fixed_count 属性

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

locust -f locust_file.py WebUser MobileUser

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

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_app1(self):
        self.client.get("/hello")
​
class WebUser(User):
    @task
    def restart_app2(self):
        self.client.get("/world")

2.4、任务

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

2.4.1、@task 装饰器

为用户添加任务的最简单方法是使用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将有两倍的机会被选中执行测试

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

2.4.2、任务属性

定义用户任务的另一种方法是设置tasks属性，tasks属性可以是一个 Task 列表，也可以是一个< Task : int> dict

from locust import User, constant
​
def my_task(user):
    pass
​
class MyUser(User):
    tasks = [my_task]
    wait_time = constant(1)

{my_task: 3, another_task: 1}

my_task被执行的可能性是another_task的 3 倍。

2.4.2、@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 tag3，则只会执行task1、task2和task4。

2.5、event事件

Locust 带有许多事件挂钩，可用于以不同方式扩展 Locust

2.5.1、对请求的测试前置test_start 、后置处理test_stop

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

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

2.6、HttpUser 类

比User类更常用，因为它添加了一个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/")

2.6.1、client属性/HttpSession

client属性是HttpSession类的一个实例，HttpSession是requests.Session的子类，requests就是常用来做接口测试的库。HttpSession没有对requests.Session做什么改动，主要是传递请求结果给Locust，比如success/fail，response time，response length，name。

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

由于requests.Session会暂存cookie，所以示例中登录/login请求后可以继续请求/my-profile。

2.6.2、断言响应结果

可以使用with语句和catch_response参数对响应结果进行断言：

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

或者直接抛出异常：

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

2.6.3、name参数

name参数用于把不同api按同一分组进行统计，比如：

for i in range(10):
    self.client.get("/blog?id=%i" % i, name="/blog?id=[id]")

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

2.6.4、HTTP代理

Locust默认设置了requests.Session的trust_env为False，不查找代理，以提高运行性能。如果需要可以设置locust_instance.client.trust_env为True。