Locust介绍 Locust是一款使用
Python开发的开源
性能
测试工具，支持分布式，可在多台主机上对系统持续发送请求，与
Jmeter、
LoadRunner的等压测工具使用线程的方式模拟用户请求不同，Locust是使用协程的方式模拟用户请求，协程的上下文切换是由自己控制，当一个协程执行完成后会主动让出，让另一个协程开始执行，切换是在用户态完成的，而线程切换是受系统控制，是在用户态与内核态之间切换，所以协程上下文切换的代价远比线程切换的代价小的多，因此Locust可以达到更高数量级的并发。
Locust安装 需要先安装python3.6以上版本，然后再安装Locust。 pip install locust # 安装locust locust -V # 查看版本
Locust文件编写 locust是通过编写python脚本进行测试的，直接看代码吧！对于单接口场景，如下所示： TaskSet类和SequentialTaskSet类实现了虚拟用户所执行任务的调度算法，包括挑选任务、执行任务、规划执行顺序等等；HttpUser中有一个用于发送http请求的client属性，在发送请求前需先调用HttpUser类；task装饰器是为了给用户添加任务，只有带有task装饰器的方法才会发起请求；constant是设置固定等待时间。 # OPLogin.py文件 import re from locust import HttpUser,task,TaskSet,constant # openstack登录 class MyTask(TaskSet): # on_start/on_stop是前置和后置操作方法，与jmeter中的setup/teardown类似，开始前后执行一次 def on_start(self): self.op_url = "/dashboard/auth/login/" # 访问登录页面 res = self.client.get(url=self.op_url, name="获取token") # 使用正则表达式在响应文档中提取动态token csrf = re.findall('name="csrfmiddlewaretoken" value="(.+(?=">))', res.text) # 将列表转为字符串，使用join方法将非字符串的格式内容都去掉，把列表中的内容都取出来，用空字符进行连接 self.token = "".join(csrf) self.cookie = res.cookies print("********** 开始执行
性能测试 **********") def on_stop(self): print("********** 测试结束 **********") @task # @task装饰器的作用是将opLogin标记为可被调用的方法 def opLogin(self): # 登录参数 data = { "csrfmiddlewaretoken": self.token, # 调用on_start中获取的token "region": "default", "username": "admin", "password": "0565317d22c54a97" } # 发送登录请求，url与on_start中的一样，直接调用 resp = self.client.request(method="post",url=self.op_url,data=data,name="登录openstack") try: # 断言，判断响应文本中是否包含指定内容，包含则返回“成功”，否则返回“失败”并打印异常 assert "Instance Overview - OpenStack Dashboard" in resp.text print("登录成功") except Exception as e: print("登录失败，未找到【Instance Overview - OpenStack Dashboard】",e) class OPlogin(HttpUser): tasks = [MyTask] # 要执行的任务 host = "http://192.166.66.33" # 被测网站域名 wait_time = constant(3) # 请求间隔时间，每次请求间隔固定3s
对于多
接口测试，也是同样的操作，在每个方法上加上task装饰器即可，task装饰器上可添加权重参数，如下所示： # CreateRES.py文件 import re from locust import HttpUser,task,TaskSet class MyTask(TaskSet): # 由于未登录无法访问创建资源，所以先直接将获取token和登录都放到on_start中 def on_start(self): self.op_url = "/dashboard/auth/login/" # 发送获取token请求 res = self.client.get(url=self.op_url, name="获取token") csrf = re.findall('name="csrfmiddlewaretoken" value="(.+(?=">))', res.text) self.token = "".join(csrf) self.cookie = res.cookies data = { "csrfmiddlewaretoken": self.token, "region": "default", "username": "admin", "password": "0565317d22c54a97" } # 发送登录请求 self.client.request(method="post",url=self.op_url,data=data,name="登录openstack") print("********** 开始执行性能测试 **********") @task(6) # (6)是可选权重参数，自定义权重占比为6 def CreateVolume(self): vol_url = "/dashboard/project/volumes/create/" vol_data = { "csrfmiddlewaretoken":self.token, # 调用on_start中获取的token "name":"dyd-volume", "volume_source_type":"no_source_type", "type":"iscsi", "size":"1", "availability_zone":"nova" } # 发送创建云硬盘的请求，需有cookie才能创建成功，在on_start中获取cookie后调用 self.client.request(method="post",url=vol_url,data=vol_data,cookies=self.cookie,name="创建云硬盘") @task(3) # 权重占比为3 def CreateSecGroup(self): csecg_url = "/dashboard/project/security_groups/create/" csecg_data = {"csrfmiddlewaretoken":self.token,"name":"dyd-SecG"} self.client.request(method="post",url=csecg_url,data=csecg_data,name="创建安全组") @task def SecGList(self): secg_url = "/dashboard/project/security_groups/" self.client.request(method="get",url=secg_url,name="查看安全组列表") class OPlogin(HttpUser): tasks = [MyTask] host = "http://192.166.66.33"
对于业务流的测试，需要使用SequentialTaskSet类，可以按代码顺序由上到下执行带有task装饰器的任务。 # Operflow.py文件 import re from locust import HttpUser,task,between,SequentialTaskSet # openstack中创建云硬盘业务流性能测试 class MyTask(SequentialTaskSet): @task # 步骤一，先获取token def get_token(self): self.op_url = "/dashboard/auth/login/" res = self.client.get(url=self.op_url, name="获取token") # 使用正则表达式在响应文档中提取动态token csrf = re.findall('name="csrfmiddlewaretoken" value="(.+(?=">))', res.text) # 将列表转为字符串，使用join方法将非字符串的格式内容都去掉，把列表中的内容都取出来，用空字符进行连接 self.token = "".join(csrf) self.cookie = res.cookies print("********** 开始执行性能测试 **********") @task # 步骤二，拿到token后登录到平台 def opLogin(self): data = { "csrfmiddlewaretoken": self.token, # 调用on_start中获取的token "region": "default", "username": "admin", "password": "0565317d22c54a97" } # 发送登录请求，url与on_start中的一样，直接调用 resp = self.client.request(method="post",url=self.op_url,data=data,name="登录openstack") try: # 断言，判断响应文本中是否包含指定内容，包含则返回“成功”，否则返回“失败”并打印异常 assert "Instance Overview - OpenStack Dashboard" in resp.text print("登录成功") except Exception as e: print("登录失败，未找到【Instance Overview - OpenStack Dashboard】",e) @task # 步骤三，登录平台后创建资源 def CreateVolume(self): vol_url = "/dashboard/project/volumes/create/" vol_data = { "csrfmiddlewaretoken":self.token, "name":"dyd-volume", "volume_source_type":"no_source_type", "type":"iscsi", "size":"1", "availability_zone":"nova" } # 发送创建云硬盘的请求 self.client.request(method="post",url=vol_url,data=vol_data,cookies=self.cookie,name="创建云硬盘") class OPlogin(HttpUser): tasks = [MyTask] host = "http://192.166.66.33" wait_time = between(0.5,2) # 请求间隔时间，会在0.5~2s之间产生一个随机数作为思考时间
也可以只调用HttpUser类，通常代码量少时使用，tag装饰器是为了标记任务，可以执行指定带有某个标记的任务。 # jpress.py文件 from locust import HttpUser, task, tag class MyJPress(HttpUser): host = "http://192.166.66.22:8080" @tag("login") @task def dologin(self): login_url = "/user/doLogin" data = {"user":"admin","pwd":"123456"} self.client.request(method="post",data=data,url=login_url,name="登录JPress") @tag("details","info") @task def lkarticle(self): url = "/article/2" self.client.request(method="get",url=url,name="查看
文章详情")
断言 上文中登录通过查找响应结果中是否包含关键字判断是否正确，还可以通过响应码、响应状态、响应时间等方法进行断言。 from locust import HttpUser, task, tag class MyJPress(HttpUser): host = "http://192.166.66.22:8080" @task def dologin(self): login_url = "/user/doLogin" data = {"user":"admin","pwd":"123456"} resp = self.client.request(method="post",data=data,url=login_url,name="登录JPress") if resp.status_code != 200: # 通过状态码判断 print("访问出错啦！暂时无法登录") elif resp.json()["state"] == "ok": # 通过响应状态判断 print("登录成功") elif resp.elapsed.total_seconds() > 1: # 通过响应时长判断 print("请求响应时间过长") @task def lkarticle(self): url = "/article/2" # 使用catch_response参数和with语句判断请求结果 with self.client.request(method="get",url=url,name="查看文章详情",catch_response=True) as res: if "小测试" in res.text: res.success() else: res.failure("无法查看文章详情！")
参数化 可借助第三方库mimesis或faker进行参数化，可自动生成账号、密码、邮箱、文本、地址、日期等。 pip install mimesis # 安装mimesis库 from locust import HttpUser, task, tag from mimesis import Text # 导入mimesis库中的Text模块 class MyJPress(HttpUser): host = "http://192.166.66.22:8080" @task def dologin(self): login_url = "/user/doLogin" data = {"user":"admin","pwd":"123456"} resp = self.client.request(method="post",data=data,url=login_url,name="登录JPress") @task def writeArt(self): csrf_url = "/ucenter" csrf = self.client.request(method="get", url=csrf_url, name="获取csrf_token") self.token = csrf.cookies["csrf_token"] # 提取csrf_token值 war_url = "/ucenter/article/doWriteSave" art = Text("zh") data = {"article.edit_mode":"markdown", "article.title": art.color() + art.level() +"……", # 使用mimesis生成标题 "article.content":art.text(), # 使用mimesis生成文章内容 "csrf_token": self.token } resp = self.client.request(method="post",url=war_url,data=data,name="投稿文章")
Locust文件执行
GUI模式 单击压测，-f表示要执行的文件，--tags表示执行带有指定标签的任务。 locust -f 路径/文件名.py # 执行文件中所有待@task装饰器的任务 locust -f 路径/文件名.py --tags login info # 执行文件中标签为login、info的任务
之后在
浏览器中输入IP:8089访问Web页面，输入总用户数和每秒增加用户数，因为代码中已经定义Host，Web页面会自动带入，若未定义可在命令中带上--host参数，或者直接在Web页面输入。
分布式压测，使用多台主机时需指明主从机器，--master表示作为主机使用，--worker表示作为从机使用，--master-host与--worker一起使用，设置主机的主机名或IP，分布式压测每台机器上都要需要安装locust，而且需要将文件拷贝至所用到的每台机器上。 locust -f 路径/文件名.py --master # 在主机上执行此命令，Web界面会在此主机上运行 locust -f 路径/文件名.py --worker --master-host=192.166.66.22 # 在每台从机上执行此命令，指定主机IP 
非GUI模式 单机压测，--headless表示使用没有UI界面的方式运行，-u表示总用户数，-r表示每秒增加的用户数，-t表示测试运行时间，--html是输出测试报告，也可以使用--csv导出csv格式的测试结果，会导出4个结果文件，-t、--html、--csv都是可选参数。 locust -f 路径/文件名.py --headless -u 30 -r 2 -t 1h30m --html=路径/报告名.html
分布式压测，非GUI模式使用以下命令，--expect-workers表示等待几个从机连接后再开始测试，从机个数达到要求后会立即开始执行。 locust -f 路径/文件名.py --headless -u 10 -r 2 -t 5s --master --expect-workers=2 # 在主机上执行，可带上报告参数 locust -f 路径/文件名.py --headless --worker --master-host=192.166.66.22 # 在每台从机上执行此命令 
监控工具 基本性能指标：响应时间、吞吐量、资源利用率，使用Locust自带的监控功能即可。 也可以使用系统自带的监控工具：vmstat、top，或nmon、dstat，后两项需要使用yum命令安装。 还可以使用Prometheus+Grafana实现更炫酷的监控图表，如下所示，整个压测过程系统性能变化会以图表形式展示。
通常一个正常系统应随着压力增大CPU的利用率也应不断升高，所以性能测试首先关注的是CPU，CPU上不去就意味着压力上不去，压力上不去又意味着无法测试系统的处理能力，继续压测并没有什么意义。单台机器的CPU处理能力是有限的，所以通常采用分布式的方式进行性能测试，从上文的报告就可以看出单台机器与分布式测试的RPS是有很大差距的，只有压力上去之后测试的结果才更准确，更容易确定系统瓶颈，常见的系统瓶颈有线程阻塞、
数据库加锁、磁盘IO阻塞等。
本文内容不用于商业目的，如涉及知识产权问题，请权利人联系51Testing小编(021-64471599-8017)，我们将立即处理
权威发布，测试选择不纠结！第15届软件测试行业报告，直击行业发展，把握未来方向！
