Locust 是一个纯python实现的可以进行分布式负载测试的工具。实际上它是基于requests库，但是它摒弃进程和线程，采用的是协程（Gevent），大大减少开销，所以它可以在单台压力机上产生数以千计的请求数，比起Jmeter的500上限高了不少

1.安装

支持3.6及以上版本

pip install locust -i https://pypi.tuna.tsinghua.edu.cn/simple

2.简单使用

from locust import HttpUser, task, between


# 相当于模拟一个用户
class WebUser(HttpUser):
    # 服务器的地址
    host = 'http://localhost:8000'

    # 下一个任务执行之前等待的时间，用于模式用户的思考时间
    wait_time = between(3, 5)  # 这里使用随机 3，5 秒钟

    @task(1)
    def info(self):
        print("获取用户信息")
        self.client.get('/info/')

    @task(10)
    def addresses(self):
        print("获取用户地址")
        self.client.get('/addresses/')

    # 测试之前会执行（前置）
    def on_start(self):
        print('用户登录')
        auth = {
    'username': 'root', 'password': 'root1234'}
        self.client.post('/login/', json=auth)

    # 测试之后会执行（后置）
    def on_stop(self):
        print('用户退出')
        self.client.delete('/logout/')

说明：

1.继承HttpUser，可以使用HttpUser.client属性发起请求，会自动状态保持

2.请求的域名（或ip）及端口写在host属性，后面的访问都只需写路径即可

3.通过 @task 装饰一个方法，可以使用权重参数，权重越大执行的几率越大

4.wait_time()表示执行下一个任务之前等待的时间

5.on_start()、on_stop()是测试开始和结束的时候会执行，类似前后置

3.启动测试

我们cd到 locustfile.py 文件所在目录，然后直接使用locust 命令，它会自动执行该文件。如果不进入，可以使用-f参数指定，比如说locust -f xxx/locustfile.py

服务器运行起来之后，会给出一个url，我们使用浏览器打开http://localhost:8089，就能看到测试界面了

我们设置好模拟得用户数和每秒启动的用户数，然后开始测试。可以看到测试的结果数据

补充：
我们也可以使用命令行参数设置无界面启动，locust -f locustfile.py --headless -u 100 -r 10 -t 10s
–headless表示不启动界面，-u表示模拟的用户数，-r表示每秒启动的用户数，-t表示测试的时间(单位可以是s、m、h)

4.分布式

因为单台电脑的性能有限，还是不能同时发起大批量的压力测试，可以考虑使用主从分布式结构

# 启动无界面主进程
locust -f locustfile.py --master --headless --master-bind-host=127.0.0.1 --master-bind-port=8090 -t 15s -u 5 -r 1

# 启动无界面从进程
locust -f locustfile.py --worker --master-host=127.0.0.1 --master-port=8090 --headless

参数是指定主进程的ip和端口

5.TaskSet

TaskSet是指动作集合，就是将某个功能整合在一起，能够完成一系列动作

import uuid

from locust import HttpUser, task, between, TaskSet


class RegisterTaskSet(TaskSet):
    # 用户名是否重复
    @task
    def username_count(self):
        self.client.get('/usernames/zhansgan/count/')

    # 手机号是否重复
    @task
    def mobile_count(self):
        self.client.get('/mobiles/13977778888/count/')

    # 获取图形验证码
    @task
    def image_codes(self):
        self.client.get(f'/image_codes/uuid-uuid-uid/')

    @task
    def stop(self):
        print('退出当前动作集合')
        self.interrupt()


class UserLoginTaskSet(TaskSet):
    # 嵌套用户注册
    tasks = [RegisterTaskSet]

    @task
    def username_login(self):
        auth = {
    'username': 'root', 'password': 'root1234'}
        self.client.post('/login/', json=auth)

    @task
    def stop(self):
        print('退出当前动作集合')
        self.interrupt()


# 模拟用户
class WebUser(HttpUser):
    # 包含用户登录
    tasks = [UserLoginTaskSet]
    # 服务器的地址
    host = 'http://localhost:8000'

    wait_time = between(3, 5)  # 这里使用随机 3，5 秒钟

    @task
    def search(self):
        self.client.get('/search/')

6.SequentialTaskSet

执行上面的代码我们发现执行是无序的，如果需要有序，我们可以继承SequentialTaskSet，它与TaskSet的区别就是，它会按照从上往下的顺序执行测试，其他的没啥区别。哦，别忘了，继承SequentialTaskSet 之后task 装饰器的权重会无效

7.断言

如果需要断言，我们可以使用上下文管理器来管理返回的响应，不过需要指定catch_response=True参数才能获取上下文管理器

from locust import HttpUser, task, between

class WebUser(HttpUser):
    wait_time = between(1, 5)
    host = 'http://localhost:8000'

    @task
    def info(self):
        with self.client.get('/info/', catch_response=True) as resp:
            res = resp.json()
            if res['code'] != 0:
                resp.failure(res['errmsg'])