【测试】Locust的安装、使用、分布式启动、TaskSet、SequentialTaskSet、断言

迈不过友情╰ 2022-11-20 11:53 98阅读 0赞

**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`，就能看到测试界面了

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center]

我们设置好模拟得用户数和每秒启动的用户数，然后开始测试。可以看到测试的结果数据  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center 1]

**补充：**  
我们也可以使用命令行参数设置无界面启动，`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'])

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/973899646ffc40e7aeccffa1226bbd6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201028203256945.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5MTQ3Mjk5,size_16,color_FFFFFF,t_70#pic_center