Locust性能测试入门案例及分布式压测

Posted 2022-12-18

前言

大家好，我是洋子。性能测试已经成为作为测试工程师/测试开发工程师一项重要的专项能力，在抢红包，活动秒杀这种短时间内流量突增的场景，已或者是健康宝这种使用用户超级多的场景，我们均需要进行压力测试，确保服务稳定可用，随着近年来测试行业的发展，业务架构越发复杂，性能测试在非功能测试场景下也成为了必要的质量保障手段，学会它早日升职加薪

工欲善其事，必先利其器，想要开展性能测试的前提，我们必须有可以利用的性能测试工具，如果所在中小型公司（公司只有我一个测试，老板还让我压测），在没有公司压测平台的辅助下，我们就得自己寻找一款性能测试工具，作为发压器帮助我们进行性能测试

对于服务端的性能测试工具，从2012年到2022年近十年的时间，有一些新的工具横空出世，如​​k6​​​、​​Vegata​​​、​​Gatling​​​、​​WebLOAD​​​等，但主流的目前依旧还是​​JMeter​​​、​​Loadrunner​​​、​​Locust​​这三款服务端性能测试工具各占一席之地

这篇文章主要介绍​​Locust​​​的使用方法，​​Locust​​​是基于​​Python​​的开源性能测试工具，用于压力测试Web应用程序。它允许自定义用户行为模型（压测任务），并使用虚拟用户来模拟真实用户访问你的应用程序，Locust借助gevent库对协程的支持，以greenlet来实现对用户的模拟，在相同配置下，Locust能支持的并发用户数相比用多线程模拟用户的Jmeter更多

Locust的使用方法非常简单，能够通过编写Python性能测试脚本，快速进行性能测试，同时也支持分布式压测，使用多台机器模拟超高并发下的压测场景

Locust还可以提供有关应用程序性能的实时报告，帮助你了解如何提高应用程序的吞吐量和响应时间

Locust经过多次迭代，老版本​​1.0​​​以前和最新版本​​2.13.2​​​在安装和使用方法上略有不同，下面的文章所使用的Locust 版本为​​2022年12月9日​​​发布的​​2.13.2​​，测试工具还是建议大家使用最新版本，修复已知的问题，性能方面更加稳定

下载安装Locust

前面说到Locust是基于Python，那运行Locust时自然就需要依赖Python环境，先安装好Python环境

（1）打开Python的官网，下载安装好​​Python 3.x​​环境

https://www.python.org/downloads/

Windows电脑，可以下载​​Windows installer​​ 安装程序，根据提示一键安装

MAC电脑，可以下载​​macOS 64-bit universal2 installer​​安装程序，根据提示一键安装

Linux系统，可以下载​​Gzipper source tarball​​ ，即gzip格式的压缩包，使用FTP上传到Linux服务器上，然后参考下面两篇安装教程进行安装

https://www.linuxprobe.com/linux-centos7-python3.html

（2）继续安装​​Locust​​​，可以借助Python的pip安装工具，在命令行终端输入安装命令，MAC电脑使用pip工具时，命令前面还需要使用​​sudo​​获取权限

# windows电脑 pip安装命令
pip install locust 
# mac电脑 pip安装命令
sudo pip install locust 

# 注意1.0以下老版本安装命令为 pip install locustio

安装完毕后，可以使用​​locust --help​​来验证是否安装成功

Locust性能测试Demo

接下来利用Locust编写一个简单的Python性能测试脚本，来快速熟悉Locust如何使用

以下性能测试脚本我们命名为​​locust_file.py​​，文件名字可以自己随意起

from locust import HttpUser, TaskSet, task, between


class UserBehavior(TaskSet):

    def on_start(self):
        """
        Called when a User starts executing this TaskSet
        """
        pass

    def on_stop(self):
        """
        Called when a User stops executing this TaskSet. E.g. when TaskSet.interrupt() is called
        or when the User is killed
        """
        pass

    @task(1)
    def baidu_homepage(self):
        self.client.get("/")

    @task(2)
    def baidu_search(self):

        with self.client.get("/s?wd=chatgpt", catch_response=True) as response:

            if response.status_code != 200:
                response.success()
            else:
                response.failure("请求失败")

            if response.json()["param1"] != "xx":
                response.failure("请求失败，No data")

            # 支持断言检查 assert response.json()[param1] == 100,"数据返回错误"

        # Post请求例子 res = self.client.post("/login", json="username": "foo", "password": "bar")


class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    host = "https://www.baidu.com"

    # 等待响应时间在3-7s内，超时处理策略
    wait_time = between(3, 7)

解读一下这个脚本的内容，一共有两个类组成，分别是​​UserBehavior​​​类和​​TaskSet​​类

UserBehavior类

​​UserBehavior​​​类继承于​​TaskSet​​类，这个类的作用是定义压测相关的任务（说白了就是我们要压测的接口），以及压测过程当中的前后操作

​​UserBehavior​​​类定义了一组任务，在本例中有两个任务，一个是访问根路径​​/​​​，另外一个是访问​​/s​​​路径。用​​@task()​​​ 装饰的方法为一个任务，​​1​​表示一个Locust实例被挑选执行的权重，数值越大，执行频率越高。在本例子中任务​​baidu_homepage​​​的权重就小于任务​​baidu_search​​

​​UserBehavior​​​类的父类​​TaskSet​​​类当中有个属性为​​self.client​​​ 是 ​​HttpSession​​​ 的一个实例，可用于向我们要进行负载测试的目标系统发出 HTTP请求，支持发送GET、POST等HTTP请求，就跟使用Python里面requests库用法一样，因为​​UserBehavior​​​类是​​TaskSet​​​类的子类，所以可以直接使用父类的属性​​self.client​​

我们最好再对接口返回的响应结果状态以及返回的结果进行逻辑验证，如响应状态码不是200，接口返回的JSON串（假定该接口返回响应是JSON格式）当中的某个字段不符合预期，这两种情况就判定为请求失败，也可以实际接口的返回数据自行修改请求成功与失败的判定逻辑

with self.client.get("/s?wd=chatgpt", catch_response=True) as response:

  if response.status_code != 200:
    response.failure("请求失败")
  if response.json()["param1"] != "xx":
    response.failure("请求失败，No data")

返回结果验证的目的是为了避免压测时接口因传参等错误导致接口返回值不正常，及时感知接口的请求情况，别到时候因为自己的性能测试脚本当中接口定义参数有问题，压完了才发现接口一直在报错，无法起到真正的压测效果

​​on_start​​​方法可以定义压测前的前置步骤，如登陆操作获取Cookie，​​on_stop​​方法则可定义压测完成的后置处理，如删除压测数据

WebsiteUser类

​​WebsiteUser​​​类继承于​​HttpUser​​​类，定义了一个模拟用户，它会向设置的host地址(这里设置的是百度官网)，发起HTTP请求，并执行​​UserBehavior​​​中定义的任务。​​UserBehavior​​​类中定义了 ​​min_wait​​​和​​max_wait​​属性分别表示模拟用户在执行任务之间等待的最小和最大时间

locust运行命令

要运行这个locust实例，可以在命令行中使用以下命令：

locust -f locust_file.py

要是不想在脚本当中写死host地址，我们可以在​​locust_file.py​​​性能测试脚本当中可以去掉 ​​WebsiteUser​​​类的​​host​​字段，这样就可以在启动locust时灵活指定要压测的host地址

locust -f locust_file.py --host=http://example.com

该命令会在本地启动locust的Web服务，然后你就可以在浏览器中访问​​http://0.0.0.0:8089​​，使用locust的Web界面来控制模拟用户的数量和执行任务

接下来进行性能测试前的配置，设置好 并发的用户数（number of users），以及每秒产生（启动）的用户数（Spawn rate），要压测的Host地址也能进行修改。设置完成后，点击Start swarming开始压测

在压测过程当中，可以实时监控压测的状态，Requests请求总数，Fails失败总数，99%和90%的请求响应时间，Average平均响应时间等指标

细心的同学能发现Requests数据项有点意思，压测时我们设置了两个任务，一个是访问根路径​​/​​​，另一个是访问​​/s​​，为什么同时压测两者的Requests数据是1:2不是1:1呢，这是因为我们设置的task权重，一个权重是1，另外一个权重是2，权重为2的任务请求会更加频繁，刚好就是1:2的关系同时支持查看曲线图，有一个值得注意的性能指标RPS（Requests Per Second），RPS和QPS（Queries Per Second）等效，指每秒能处理完的请求数目

做过性能测试的小伙伴可能还会想到一个指标叫TPS（Transactions Per Second），每秒处理的事务数目。一个事务是指一个客户机向服务器发送请求然后服务器做出反应的过程。客户机在发送请求时开始计时，收到服务器响应后结束计时，以此来计算使用的时间和完成的事务个数，最终利用这些信息作出的评估分

TPS与QPS之间的区别

（1）若在一秒内，用户请求了百度首页并看到了首页全貌，这样就完成了一个事务（TPS=1），但其实向服务端发起了N多次HTTP请求，或者是存在只发了一次HTTP请求某个接口，但该接口下游还请求了其他接口（QPS=N），这两种情况属于一个事务包含了多个HTTP请求，此时TPS不等于QPS
（2）若在一秒内，我们请求一个查询接口，且这个接口内部不会再去请求其它接口，即一个事务只有1个HTTP请求，此时TPS等于QPS

压测完毕后可以点击Download Data下载压测报告

Locust分布式压测

Locust 默认情况下是以单进程模式运行，在这种模式下，所有的虚拟并发用户均运行在单个Python进程中，
由于单进程的原因，并不能完全发挥压力机所有CPU处理器的能力

Locust还支持多进程分布式压测模式，可以在同一台计算机或者多台计算机当中进行分布式压测

不管是单机多进程分布式压测还是多机分布式压测，运行方式都是先运行一个​​master​​​节点负责分发和统计，再启动​​work​​节点负责并发执行，如果是使用多台机器压测，需要先在所有机器上安装好locust，以及部署好完全一致的性能测试脚本，注意work节点和master节点需要在同一局域网内，能够互相通信

master节点启动命令

locust -f locust_file.py --master --master-bind-port=8089 (端口可以自己改)

work节点启动命令

locust -f locust_file.py --worker --master-host=<locust_master_machine_ip> --master-port=8089 (端口与master机器保持一致）

<locust_master_machine_ip>填写的是机器的网卡IP，我这里同时启动了两个work节点，打开Web页面，可以看到Workers的数量变成为2。在单机上进行分布式压测，Web访问页面为​​http://localhost:8089/​​

点击New test即可开始分布式压测

有小伙伴可能会说了，如果work节点数量非常多，手动部署推送性能脚本到服务器上好低效，有没有办法自动部署呢？

在单机分布式压测的情况下，我们可以编写一键启动shell脚本，同时启动locust master与work节点

echo "start master..."
nohup locust -f locust_file.py --master --master-bind-port=8089 > main.log 2>&1 &

workerNum=2
echo "start worker, size=$workerNum..."
for i in $( seq 1 $workerNum)
do
  nohup locust -f locust_file.py --worker --master-host=<locust_master_machine_ip> --master-port=8089  worker$i.log 2>&1 &
  echo "output worker$i.log"
done

echo "end..."

如果在多台机器上要同时启动压测脚本或者一键推送修改后的压测脚本，则需要自动登录远程登陆主机，​​scp​​命令推送文件，可以使用shell中except脚本实现，具体代码逻辑这里不再展开

在查找资料的时候发现网上有个大佬编写Python的脚本也实现了多台机器测性能测试脚本自动部署

最后一小点补充，locust分布式压测通信利用到​​ZMQ (ZeroMQ）​​网络通信库，ZMQ提供了在进程内、进程间、TCP和多播等各种传输中传递原子消息的套接字（socket），支持多种传输（TCP、进程内、进程间、多播、WebSocket等）上的通用消息传递模式

因此需要使用分布式压测依赖​​pyzmq​​​库，在我们使用​​pip install locust​​安装好locust会自动安装此库

附录

1. 关于Python协程官方文档：https://docs.python.org/zh-cn/3.7/library/asyncio-task.html
2. locust官方文档：https://docs.locust.io/en/stable/index.html
3. 词表的多种使用脚本参考：https://debugtalk.com/post/head-first-locust-advanced-script/
4. 关于locust的命令其他使用，可以​​locust --help​​查看
5. 
   

结束语

本文为大家介绍了用Locust编写Python性能测试脚本，以及如何使用分布式压测，不得不说使用Locust做性能测试真的很方便，当然测试工具只是性能测试当中的小小一环，可以从下方表格获取到性能测试各流程

流程阶段	做法
场景分析	识别瓶颈，建立数据模型，确定测试纬度
测试计划	执行测试方案和测试计划，指导测试执行
环境规划	规划性能测试环境（生产环境还是线下环境），避免环境瓶颈
数据 & 工具准备	测试工具使用，测试数据脚本（若有），统计脚本（若有）
指标获取 & 统计	执行测试 & 获取监控的性能指标数据
报告分析	生成报告，反馈数据，分析数据，便于后续性能调优