【Python】Python之locust压测教程+从0到1demo:高阶轻量级压测实战+常用参数详解(2)

文章目录

  • 一、主流性能测试工具对比
  • 二、 Locust性能测试工具介绍
  • 简介
  • 线程和协程的区别
  • 为什么选择Locust?
  • 三、创建一个登陆接口
  • flask接口文档
  • 请求参数
  • 响应参数
  • 响应示例
  • 四、测试多个接口性能
  • 4.1 导入必要模块
  • 4.2 接口测试demo
  • 4.3 启动测试用例
  • 五、深入解析Locust核心参数
  • 5.1 用户类参数
  • 5.1.1 `HttpUser`与`User`
  • 5.1.2 `tasks`
  • 5.1.3 `wait_time`
  • 5.2 命令行参数
  • 5.2.1 `-f` / `–locustfile`
  • 5.2.2 `-u` / `–users`
  • 5.2.3 `-r` / `–spawn-rate`
  • 5.2.4 `–headless`
  • 5.2.5 `–run-time`
  • 5.2.6 `–csv`
  • 5.3 SLAs(服务水平协议)
  • 上一篇章我们讲的是如何把接口测试用例,转换成locust性能测试用例,那么本篇文章我们就深入了解locust的高阶实战demo,以及相关参数的详解。

    【Python】Python之locust压测教程+从0到1demo:基础轻量级压测实战(1)

    【Python】Python之locust压测教程+从0到1demo:高阶轻量级压测实战+常用参数详解(2)

    一、主流性能测试工具对比

    工具 是否收费 语言 并发机制 单机并发能力 场景压测 分布式
    Locust 免费 Python 协程 支持 支持
    Jmeter 免费 Java 线程 支持 支持
    Loadrunner 收费 C 进程/线程 支持 支持

    二、 Locust性能测试工具介绍

    简介

    Locust是一款易于使用的分布式负载测试工具,一个locust节点就可以在一个进程中支持数千并发用户,基于事件,通过gevent使用轻量级执行单元。

    线程和协程的区别

  • 一个线程可包含多个协程
  • 线程和进程是同步机制,协程是异步
  • 线程的切换由操作系统调度,协程由用户自己进行调度
  • 协程相较于线程更加轻量,资源消耗更低
  • 为什么选择Locust?

  • 基于协程,低成本实现更多并发
  • 有第三方插件,易于扩展
  • 支持分布式、高并发能力
  • 三、创建一个登陆接口

    如果有接口的用户可以忽略这一步

    from flask import Flask, make_response, request
    
    app = Flask(__name__)
    
    @app.route('/login', methods=['GET', 'POST'])
    def login():
        # 尝试从GET请求中获取参数
        username = request.args.get('username', type=str)
        password = request.args.get('password', type=int)
    
        # 如果GET请求中没有参数,则尝试从POST请求中获取参数
        if username is None or password is None:
            username = request.form.get('username', type=str)
            password = request.form.get('password', type=int)
    
        # 验证用户名和密码
        if username == 'admin' and password == 123:
            response_data = {
                'code': 0,
                'msg': "OK",
                'data': {
                    'test': f"{request.method.lower()}请求测试页面"
                }
            }
        else:
            response_data = {
                'code': 999,
                'data': f"无效的用户名或密码: {username} {password}"
            }
    
        return make_response(response_data)
    
    if __name__ == '__main__':
        app.run(host='0.0.0.0', port=8080, debug=False, threaded=True)
    
    

    当启动后我们可以看到一个url,这就我们要测试demo的接口。

    通过运行,我们可以整理出一个接口文档如下:

    flask接口文档

  • URL: http://127.0.0.1:8080
  • 接口路径: /login
  • 请求方法: GETPOST
  • 请求参数
    参数名 类型 必填 描述
    username string 用户名
    password int 密码(整数类型)
    响应参数
    参数名 类型 描述
    code int 响应状态码,0表示成功,999表示失败
    msg string 响应消息,成功时为"OK",失败时为错误信息
    data object 成功时包含测试页面信息,失败时为错误描述
    响应示例
  • get/post成功响应示例:
  • {
        "code": 0,
        "msg": "OK",
        "data": {
            "test": "get请求测试页面"  # 或者"post请求测试页面"
        }
    }
    
  • 失败响应示例:
  • {
        "code": 999,
        "data": "无效的用户名或密码: admin 123"
    }
    

    四、测试多个接口性能

    4.1 导入必要模块

    from locust import HttpUser, task, between
    
  • HttpUser:Locust的核心类,用于模拟用户行为。
  • task:装饰器,用于标记测试任务。
  • between:用于定义任务之间的等待时间。
  • 4.2 接口测试demo

    from locust import HttpUser, task, between
    
    
    class MyUser(HttpUser):  # 1、在类中继承locust.HttpUser的子类,也就是创建这个子类
    
        # 添加locust给我们提供的函数between,里面1和2的值代表着每个task间隔1到2秒
        wait_time = between(1, 2)
    
        @task(1)  # 权重为1
        def test_login_get(self):
            resp = self.client.get('http://127.0.0.1:8080/login?username=admin&password=123')
            print(resp.json())
            assert resp.status_code == 200
    
        @task(2)  # 权重为2,执行频率更高
        def test_login_post(self):
            resp = self.client.post("http://127.0.0.1:8080/login", data={"username": "admin", "password": "123"})
            print(resp.json())
            assert resp.status_code == 200
    
  • HttpUser:Locust的核心类,用于模拟用户行为。

  • task:装饰器,用于标记测试任务。

  • between:用于定义任务之间的等待时间。

  • @task(1):定义了一个权重为1的任务,权重决定了任务的执行频率。

  • test_login_get:任务名称,这里模拟了一个GET请求,进行登录操作。

  • self.client.get(…):发送HTTP GET请求。

  • assert resp.status_code == 200:断言响应状态码为200,确保请求成功。

  • 同样,test_login_post任务的权重为2,意味着它的执行频率是test_login_get的两倍。这种权重设置可以准确模拟不同用户行为的真实分布。

    4.3 启动测试用例

    我们在控制台输入命令

    locust -f 测试文件.py --host=http://127.0.0.1:8080/
    

    这样我们就得到了一个GUI的性能测试地址,我们打开链接

    我们选择100个用户并发,每秒递增10个然后进行启动。


    这样我们就能看到在一个性能测试的页面里面,有多个接口在进行测试。并且我们设置的@task(2) 的接口数量是远超与@task(1) 的。

    这就是我们要讲的,如何进行多接口压测。

    五、深入解析Locust核心参数

    为了充分发挥Locust的性能,我们需要深入理解其核心参数及其作用。以下是Locust中常用的一些关键参数及其详细解释。

    5.1 用户类参数

    5.1.1 HttpUserUser
  • HttpUser:适用于需要发送HTTP请求的性能测试场景,内置了self.client用于发送请求。
  • User:基本用户类,不自带self.client,适用于非HTTP协议的测试。
  • 根据具体需求选择合适的用户类,可以提高测试的灵活性和效率。

    5.1.2 tasks
    tasks = {task1: weight1, task2: weight2, ...}
    

    tasks属性定义了用户在测试过程中可以执行的多个任务及其权重。权重决定了任务被执行的概率和频率。例如:

    tasks = {test_login_get: 1, test_login_post: 2}
    

    表示test_login_post的执行频率是test_login_get的两倍。

    5.1.3 wait_time
    wait_time = between(min_wait, max_wait)
    

    wait_time设置了用户在执行两个任务之间的等待时间。Locust提供了多种等待时间函数:

  • between(min, max):在minmax秒之间随机等待。
  • constant(seconds):固定等待指定秒数。
  • constant_pacing(seconds):确保每个任务之间的间隔至少为指定秒数。
  • 我们合理设置等待时间,可以更真实地模拟用户行为,避免过度负载。

    5.2 命令行参数

    Locust提供了丰富的命令行参数,用于灵活配置测试。

    5.2.1 -f / --locustfile

    指定Locust的测试脚本文件。默认情况下,Locust会寻找locustfile.py

    locust -f 测试代码.py
    
    5.2.2 -u / --users

    定义模拟的总用户数。例如,-u 100表示有100个并发用户。

    locust  -f 测试代码.py -u 100
    
    5.2.3 -r / --spawn-rate

    定义用户生成的速率,即每秒生成多少用户。例如,-r 10表示每秒生成10个用户。

    locust  -f 测试代码.py -r 10
    
    5.2.4 --headless

    以无头模式运行,不启动Web界面,适合自动化脚本和持续集成。

    locust  -f 测试代码.py --headless -u 100 -r 10
    
    5.2.5 --run-time

    定义测试的持续时间。例如,--run-time 1h表示测试持续1小时。

    locust  -f 测试代码.py --headless -u 100 -r 10 --run-time 1h
    
    5.2.6 --csv

    将测试结果保存为CSV文件,便于后续分析。

    locust -f 测试代码.py -u 100 -r 10 --csv=performance_test
    

    生成的文件将包括performance_test_stats.csvperformance_test_failures.csv等,详细记录了请求的响应时间、失败率等关键指标。

    5.3 SLAs(服务水平协议)

    服务水平协议(SLAs)是定义预期性能标准的重要工具。Locust支持特定指标设定SLAs和自动监控和警报。

    from locust import constant, HttpUser, task, between, events
    
    class MyUser(HttpUser):
        wait_time = between(1, 2)
    
        @task
        def test_login_post(self):
            resp = self.client.post("http://127.0.0.1:8080/login", data={"username": "admin", "password": "123"})
            assert resp.status_code == 200
            # 设定响应时间不超过500ms
            if resp.elapsed.total_seconds() > 0.5:
                events.request_failure.fire(
                    request_type="POST",
                    name="test_login_post",
                    response_time=resp.elapsed.total_seconds(),
                    exception="Response time exceeded SLA"
                )
    

    通过设定SLAs,就可以确保代码在规定的性能范围内运行,一旦超出,系统会自动记录失败事件,帮助你及时发现和解决问题。

    我们通过两章完整的讲述了locust的全部demo,这些知识已经足以满足我们的测试需求。喜欢的可以点赞关注,分享更多测试文章!

    作者:ChatGPT-千鑫

    物联沃分享整理
    物联沃-IOTWORD物联网 » 【Python】Python之locust压测教程+从0到1demo:高阶轻量级压测实战+常用参数详解(2)

    发表回复