locust接口压测实战

Overview

Locust工具虽然在很多年前使用过，但从未做过相关内容的总结。本次在测试过程中有试用到此工具，故而简单mark一下。

Locust是一个比较容易上手的、开源的分布式用户负载测试工具，它允许你使用Python代码来定义用户行为，并模拟数以万计的用户对网站或应用程序进行压力测试。Locust 在英文中是 蝗虫 的意思：作者的想法是在测试期间，放一大群 蝗虫 攻击你的网站。当然事先是可以用 Locust 定义每个蝗虫（或测试用户）的行为，并且通过 Web UI 实时监视围攻过程。由于其灵活性及易用性，Locust在开发者和测试工程师中相当受欢迎。

Locust 的运行原理是完全基于事件运行的，因此可以在一台计算机上支持数千个并发用户。与许多其他基于事件的应用程序相比，它不使用回调（比如 Nodejs 就是属于回调，Locust 不使用这种的逻辑）。相反，它通过 gevent 使用轻量级进程。测试你站点的每个蝗虫实际上都在其自己的进程中运行。

安装 locust

pip3 安装 locust

pip3 install locust

查看是否安装成功

locust -h

运行 locust 分布在多个进程/机器库

pip3 install pyzmq

安装 webSocket 压测库

pip3 install websocket-client

TaskSequence 类

Locust 提供了一些核心的类，使得你可以通过编写Python脚本来定义模拟用户的行为。以下是一些主要的 Locust 类及其作用：

HttpUser

这是 Locust 中最常用的类之一，用于模拟发送HTTP请求的用户。继承此类可以定义发送HTTP请求的任务，如访问网页、API调用等。HttpUser 类为你的用户任务提供了 .client 属性，通过这个属性，你可以发起 GET、POST 等请求。

TaskSet

TaskSet 允许你定义一个用户执行的任务集合。一个 TaskSet 可以包含一个或多个任务，任务之间可以有不同的权重，表示任务被执行的相对概率。TaskSet 可以被嵌套，即一个 TaskSet 中可以包含另一个 TaskSet。

task

task 是一个装饰器，用于将某个方法标记为一个任务。这个装饰器可以应用于 HttpUser 和 TaskSet 类下的方法。可以通过为装饰器设置权重参数（一个整数）来调整任务被执行的频率。

SequentialTaskSet

这个类是 TaskSet 的一个特殊版本，它会按照任务被定义的顺序来执行任务，而不是随机选择任务执行。这对于需要以特定顺序执行任务的测试非常有用，比如用户注册流程—先填写表单，然后接收验证邮件，最后验证账号。

FastHttpUser

此类与 HttpUser 类似，但使用的是更快、更轻量级的HTTP客户端（基于geventhttpclient）。FastHttpUser 适用于需要极高性能和并发量的场景，但可能不支持 HttpUser 中的某些高级HTTP功能。

between

between 是一个函数，用于为用户任务之间的等待时间设置随机时间间隔。该函数接受两个参数，分别表示最小等待时间和最大等待时间。

constant

constant 是另一个用于设置用户任务之间等待时间的函数，其提供了一个固定的等待时间。

使用示例

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(1, 5)  # 用户任务之间的等待时间为1到5秒之间的随机值

    @task(2)
    def read_item(self):
        self.client.get("/item")

    @task(1)
    def write_item(self):
        self.client.post("/item", {"data": "value"})

在这个示例中，MyUser 类继承自 HttpUser 类，代表一个HTTP用户。定义了两个任务：read_item 和 write_item，其中 read_item 的执行频率是 write_item 的两倍。

初始化方法

• setup 和 teardown 方法 setup 和 teardown 都是只能运行一次的方法。

  • 在任务开始运行之前运行setup，而在所有任务完成并且蝗虫退出后运行 teardown；这使你能够在任务开始运行之前做一些准备工作（比如创建数据库，或者打印日志 等等），并在Locust退出之前进行清理。

• on_start 和 on_stop 方法 每个虚拟用户执行操作时运行on_start方法，退出时执行on_stop方法。

• 初始化方法的执行顺序 setup > on_start > on_stop > teardown。

启动方式

常用3种启动方式，直接启动，无web页面启动和分布式启动。

直接启动

locust -f your_locust.py
（your_locust.py为执行脚本，可在编译器中直接运行该脚本）

无web页面启动

locust -f your_locust.py --no-web -c 200 -r 20 -t 1m

• （–no-web 代表不需要启动UI页面

• -c 代表需要并发的用户数

• -r 代表每秒并发的用户数

• -t 代表需要运行的时间）

分布式启动

Locust与Jmeter不同，Jmeter进行压测的时候，会一下子使用到CPU的所有核心，但是locust运行时，只会使用到CPU的一个核心，所以当单次运行无法达到想到的压力值。采用分布式运行，可以充分运用到CPU的所有内核，生成更大负载，也可以直接接入远程机，让负载值飞跃增长。

locust -f your_locust.py --master # 指定当前机器为master主机
locust -f your_locust.py --slave --master-host=10.xxx.xxx.xxx # 指定当前机器为从机并指向对应master主机

本地机运行分布式

主程采用master运行。

locust -f your_locust.py --master # 指定当前机器为master主机

与旧版locust不同，新版不能使用salve，需要采用worker运行负载，如下。

locust -f locust.py --worker

每运行一次，就会多一个负载。（最大负载为CPU核心数）

远程机结合运行分布式

主机负载同上，当运行负载机时，需要指定主机的IP。如下

locust -f your_locust.py --worker --master-host=IP

在 Locust 中，你可以通过命令行参数来指定测试的运行时间。使用 -t 或 --run-time 参数后面跟上你想要的运行时间，如 10s（10秒）、5m（5分钟）、1h（1小时）等。

例如，如果你想运行 Locust 测试5分钟，你可以在命令行中这样指定：

locust -f your_locust.py --users 100 --spawn-rate 10 -t 5m

在这个命令中：

• -f locustfile.py 指定了 Locust 测试脚本文件。

• --users 100 指定了总共模拟的用户数。

• --spawn-rate 10 指定了每秒生成用户的速率。

• -t 5m 或者 --run-time 5m 指定了测试运行的时间，即5分钟。

Locust 会在指定的时间结束后停止运行。这对于需要在持续集成/持续部署 (CI/CD) 管道中自动执行性能测试非常有用。这样你可以确保性能测试不会无限期地运行，而是在完成足够的模拟用户行为后自动停止。

代码模板

如下模板演示了如何使用 Locust 进行基本的负载测试，包括模拟多种用户行为、使用任务集和处理不同类型的HTTP请求。

from locust import HttpUser, TaskSet, task, between

# 定义用户行为
class UserTasks(TaskSet):
    # 通过装饰器设置任务，数字表示任务权重，权重越高任务被选中的概率越大
    @task(2)
    def view_main_page(self):
        # 模拟用户浏览主页
        self.client.get("/")

    @task(1)
    def view_about_page(self):
        # 模拟用户浏览关于页面
        self.client.get("/about")
    
    # 嵌套 TaskSet 示例，用于模拟具有多个步骤的用户交互流程
    @task(1)
    class PostMessage(TaskSet):
        # 步骤1: 浏览到发帖页面
        @task
        def view_post_page(self):
            self.client.get("/post")
        
        # 步骤2: 提交一个帖子
        @task
        def post_message(self):
            self.client.post("/submit", data={"message": "Hello World"})
            # 嵌套任务完成后，调用self.interrupt结束当前TaskSet，返回上一级TaskSet
            self.interrupt()

# 定义用户模拟类
class WebsiteUser(HttpUser):
    tasks = [UserTasks]  # 将用户任务关联到HttpUser
    wait_time = between(1, 5)  # 设置用户执行任务的等待时间，介于1到5秒随机

说明：

• UserTasks 类继承自 TaskSet，表示一组用户可以执行的任务。你可以在此类中定义多个任务，这些任务可以是访问某个页面、提交表单等。

• tasks 属性用于将 TaskSet 与用户类 HttpUser 关联起来。该属性可以是一个列表，包含多个 TaskSet 类。

• wait_time 定义了用户在执行完一个任务后，再执行下一个任务前的等待时间。这里使用了 between 函数，意味着等待时间将在1到5秒之间随机选择。

• @task 装饰器标记了哪些方法是任务。通过为装饰器提供参数，你可以设置任务的执行权重。未指定权重时，默认为1。

• self.client.get 和 self.client.post 方法分别用于发起GET和POST请求。HttpUser 类提供的 self.client 属性是一个 Session 对象，用于发起 HTTP 请求，并自动携带任何以前设置的cookies等。

• 嵌套 TaskSet (PostMessage) 演示了一个更复杂的用户行为，其中用户需要执行一系列有顺序的任务。使用 self.interrupt() 来退出嵌套的任务集合。

这个模板是一个Locust测试的起点，通过自定义和扩展这个模板，你可以为你的应用或服务创建各种复杂的负载测试场景。

实战

测试要求

假如产品内嵌搜索引擎，简单压测一下搜索能力。

测试代码

先手工发起对应的业务请求，摘取请求中的Header信息。比如通过Firefox Browser，摘取headers信息：

root@Gavin:~# cat header.search 
Host: search.xxxx.cn
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0
Accept: */*
Accept-Language: zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2
Accept-Encoding: gzip, deflate, br
Referer: https://search.xxxx.cn/aiSearchPage
RSC: 1
Next-Router-State-Tree: %5B%22%22%2C%7B%22children%22%3A%5B%22aiSearchPage%22%2C%7B%22children%22%3A%5B%22__PAGE__%22%2C%7B%7D%5D%7D%5D%7D%2Cnull%2Cnull%2Ctrue%5D
Next-Router-Prefetch: 1
Next-Url: /aiSearchPage
Connection: keep-alive
Sec-Fetch-Dest: empty
Sec-Fetch-Mode: cors
Sec-Fetch-Site: same-origin
Pragma: no-cache
Cache-Control: no-cache
root@Gavin:~#

保存为header.search文件，借助如下脚本，转换成JSON格式内容：

root@Gavin:~# cat change_head_to_json.py 
#!/usr/bin/env python

import os
import sys
import json


file_name = "./header.search"
head_json = "./header.json"


def file_check():
    if not os.path.exists(file_name):
        print(f"[ERROR]  Not find file : {file_name}, exit!!!")
        sys.exit(1)


def read_file_content(file_name):
    with open(file_name, 'r', encoding='utf-8') as file:
        return file.read()


def format_header_to_json():
    header_content = read_file_content(file_name)

    # 使用字典推导式将每一行转换为键值对，分割点为第一个": "序列
    headers = {line.split(": ")[0]: line.split(": ")[1] for line in header_content.strip().split("\n")}

    # 将字典转换为JSON并写入文件
    with open(head_json, "w", encoding="utf-8") as json_file:
        json.dump(headers, json_file, indent=4, ensure_ascii=False)


if __name__ == '__main__':
    format_header_to_json()

然后，编写发起业务请求的脚本，参考如下:

#!/usr/bin/env python
# -*- coding:GB2312 -*-

import random
from locust import HttpUser, task, between


class UserRun(HttpUser): 
    wait_time = between(0.1, 0.3)

    def on_start(self):
        print("Start\n")

    @task(2)
    def Agent_Search(self):
        url = 'https://search.xxxx.cn/aiSearchPage/search?q=locust%2520post%25E8%25AF%25B7%25E6%25B1%2582&rid=fwgto8n3F4e6m0nmHJ2X_&modelName=glm-4&_rsc=16qln'
        headers = {
    "Host": "search.xxxx.cn",
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0",
    "Accept": "*/*",
    "Accept-Language": "zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2",
    "Accept-Encoding": "gzip, deflate, br",
    "Referer": "https://search.xxxx.cn/aiSearchPage",
    "RSC": "1",
    "Next-Router-State-Tree": "%5B%22%22%2C%7B%22children%22%3A%5B%22aiSearchPage%22%2C%7B%22children%22%3A%5B%22__PAGE__%22%2C%7B%7D%5D%7D%5D%7D%2Cnull%2Cnull%2Ctrue%5D",
    "Next-Router-Prefetch": "1",
    "Next-Url": "/aiSearchPage",
    "Connection": "keep-alive",
    "Sec-Fetch-Dest": "empty",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Site": "same-origin",
    "Pragma": "no-cache",
    "Cache-Control": "no-cache"}

        # 发起请求，并捕获响应以手动处理
        with self.client.get(url, headers=headers, catch_response=True) as response:
            if response.status_code == 200:
                # 当状态码为200时，直接认为请求成功
                response.success()
            elif response.status_code == 404:
                # 即使是404状态，也可以手动将请求标记为成功
                response.failure('Agent search 404！')
            else:
                # 其他情况，默认为失败，无需手动处理
                response.failure('Agent search 接口失败！')

    def on_stop(self):
        print("End\n")

测试运行

C:\Users\Wang>locust -f C:\Users\Wang\Desktop\locust_test_script.py --users 100 --spawn-rate 10 -t 5m
[2024-02-02 15:43:34,563] Gavin/INFO/locust.main: Starting web interface at http://localhost:8089 (accepting connections from all network interfaces)
[2024-02-02 15:43:34,570] Gavin/INFO/locust.main: Starting Locust 2.21.0
[2024-02-02 15:49:49,212] Gavin/INFO/locust.runners: Ramping to 100 users at a rate of 10.00 per second

这个时候，打开http://localhost:8089/网页，会出现如下要配置页面：

• Number of users ：压力测试的用户总量，如上面的100

• Spawn rate：每秒增加的用户量，从10开始增加直到用户总量数停止。

• Host：被测接口的域名或IP端口地址(带http://或者https://)

设置好参数后，点击Start swarming按钮，开始测试，各个页面信息介绍如下。

各个页面介绍

Statistics

上图显示当前接口的URL，请求量，失败数，请求响应中值，平均值，当前RPS等信息。
RPS（也叫TPS）：每秒钟处理完的事务次数，一般TPS是对整个系统来讲的。一个应用系统1s能完成多少事务处理，一个事务在分布式处理中，可能会对应多个请求，对于衡量单个接口服务的处理能力，用QPS(每秒查询率)比较多。

顺便介绍几个概念：

• PV是指页面被浏览的次数，比如你打开一网页，那么这个网站的PV就算加了一次；

• TPS是每秒内的事务数，比如执行了DML操作，那么相应的TPS会增加；

• QPS是指每秒内查询次数，比如执行了SELECT操作，相应的QPS会增加。

不同的应用系统TPS，QPS是没有可对比性的。

例如：

应用A，每个select查询需要1 ms, 一个connection的话，一直不停的执行，1 s 内 可执行1000次，也就是1000 QPS；

应用B，每个select查询需要100 ms, 一个connection的话，一直不停的执行，1 s 内 可执行10次，也就是10 QPS。

上面不同系统的两个QPS是无法对比的，不能说哪个好哪个坏，满足业务要求才是王道。

Charts

Charts分为三个部分，分别是：

Total Requests per Second: 展示随着用户的增加，TPS变化数。

Response Times(ms): 展示接口从开始压测，到稳定实时的响应时间，随着压力增加，一般会升高。

Number of Users: 展示从0开始，每秒增加10人，直到总数100后不再增加。

Failures

只有在发生异常情况下，此页面才会有失败，记录失败的HTTP相关信息。

Exception

只有当测试脚本发送异常，比如HTTP请求发送失败，此处会记录异常信息，供调试代码使用。

Current Ratio

记录当前测试函数（红框中所标示）的ratio信息，100%最好，低于此值则测试预期可能不达标。

Download Data

测试数据下载模块， 提供四种类型的CSV格式的下载， 分别是：Statistics、responsetime、failures、exceptions。

Locust小结

优势

• 可编程性：Locust使用Python，这意味着你可以编写Python脚本来定义复杂的用户行为，利用Python的强大库资源。

• 轻量级且易于安装：Locust是一个相对轻量级的工具，安装过程简单方便。

• Web界面：提供一个易于使用的web界面用于启动测试、实时查看测试结果和统计信息。

• 可扩展性：支持分布式测试，你可以很容易地通过增加更多的Slave节点来模拟数十万甚至数百万的并发用户。

• 实时统计：通过Web UI，Locust能够实时展示请求统计、响应时间图表等数据，使结果分析更为直观。

• 适用性广：可以用来测试网站、API、数据库等多种服务的性能。

局限性

• 性能问题：虽然Locust本身是高效的，但它运行Python脚本，这可能不如一些编译语言（如C++或Go）编写的测试工具性能高。

• 资源消耗：在模拟非常高的并发用户数时，Locust（特别是运行Locust的机器）可能会消耗大量的CPU和内存资源。

• 学习曲线：对于那些不熟悉Python的用户来说，初学Locust可能会有一定的学习曲线。

• 对高级负载模式的支持有限：虽然Locust非常灵活，但在一些非常特定的负载测试场景（如具有复杂的用户会话管理或定制的网络协议）中，可能需要进行相当多的自定义编码。

• 社区和支持：尽管Locust有一个活跃的社区，但与某些更成熟的压力测试工具相比（如Jmeter），其用户基础和可用资源可能较少。

总的来说，Locust是一个功能强大且灵活的负载测试工具，特别适合那些熟悉Python并希望快速构建和执行实时监控负载测试的开发者和测试人员。然而，每个工具都有其优势和局限性，选择合适的工具应基于具体的测试需求和场景。