点击刷新
一、简介
locust是一款易于使用的分布式用户负载测试工具。它用于对网站（或其他系统）进行负载测试，并确定系统可以处理多少并发用户。
二、安装
pip install locustio==0.11.0
（python3.7安装最新版的locust可能会报错，无法正常安装，所以需要指定版本安装）
三、使用
推荐博文：https://blog.csdn.net/swinfans/article/details/88915176
1. 脚本
#单用户请求多个接口 from locust import HttpLocust,TaskSet,task,events import os from gevent._semaphore import Semaphore #集合点 # 实例化 all_locusts_spawned = Semaphore() all_locusts_spawned.acquire() # 创建等待方法 def on_hatch_complete(**kwargs): all_locusts_spawned.release() # 当用户加载完成时触发 events.hatch_complete += on_hatch_complete class UserBehavior(TaskSet): # 登录 def login(self): loginUrl = "/zentao/user-login.html/" h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0", "Content-Type": "application/x-www-form-urlencoded"} body = {"account": "yx", "password": "123456", "keepLogin[]": "on", "referer": "/zentao/my/"} r = self.client.post(loginUrl, data=body, headers=h) self.locust_asser("parent.location='/zentao/index.html", r) # 退出 def logout(self): logoutUrl='http://192.168.x.xx:80/zentao/user-logout.html' h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0", "Content-Type": "application/x-www-form-urlencoded"} r = self.client.get(logoutUrl,headers=h) self.locust_asser("/zentao/user-login.html", r) # 断言 def locust_asser(self, expect, result): if expect in result.text: result.success() else: result.failure('Fail.') def on_start(self): self.login() all_locusts_spawned.wait() def on_stop(self): self.logout() # 括号里面参数表示该行为挑选执行的权重，数值越大，执行频率越高，不设置默认是1 # @seq_task(2) 指定它的任务按顺序执行，括号中的数值为顺序值，数值越大，执行顺序越靠后 @task(2) def zentao_my(self): r = self.client.get("/zentao/my/") self.locust_asser("我的地盘",r) @task(1) def zentao_product(self): r = self.client.get("/zentao/product-browse.html/") self.locust_asser("需求列表",r) @task(1) def zentao_prject(self): r = self.client.get("/zentao/project/") self.locust_asser("项目首页",r) #设置性能测试 class WebsiteUser(HttpLocust): #指向一个定义了的用户行为类 task_set = UserBehavior #用户执行任务之间等待时间的最小时间，单位毫秒 min_wait = 3000 #用户执行任务之间等待时间的最大时间，单位毫秒 max_wait = 6000 if __name__ == "__main__": os.system("locust - f demo3.py --host=http://192.168.x.xx:80 --no-web --csv=example -c 1 -r 1 -t 10s")
#多用户请求一个接口 import xlrd,os,random from locust import HttpLocust,TaskSet,task,events from gevent._semaphore import Semaphore #集合点 # 实例化 all_locusts_spawned = Semaphore() all_locusts_spawned.acquire() # 创建等待方法 def on_hatch_complete(**kwargs): all_locusts_spawned.release() # 当用户加载完成时触发 events.hatch_complete += on_hatch_complete class UserBehavior(TaskSet): #登录 def login(self): #随机取出列表中的用户名和密码 userInfo=self.randomValue(self.logindata) loginUrl ="/zentao/user-login.html/" h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0","Content-Type": "application/x-www-form-urlencoded"} body = {"account": userInfo[0],"password":userInfo[1],"keepLogin[]": "on","referer": "/zentao/my/"} r = self.client.post(loginUrl, data=body, headers=h) self.locust_asser("parent.location='/zentao/index.html'",r) # 退出 def logout(self): logoutUrl='http://192.168.x.xx:80/zentao/user-logout.html' h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0", "Content-Type": "application/x-www-form-urlencoded"} r = self.client.get(logoutUrl,headers=h) self.locust_asser("/zentao/user-login.html", r) #断言 def locust_asser(self,expect,result): if expect in result.text: result.success() else: result.failure('Fail.') #取列表中的随机数 def randomValue(self,arr): r=random.randint(0,10000) index=r%len(arr) return arr[index] #读取excel，将每行数据放入一个列表，将所有列表放入一个列表形成二维列表，返回该二维列表 def read_excel(self,path): book = xlrd.open_workbook(path) sheet = book.sheets()[0] li = [] for i in range(1, sheet.nrows): li.append(sheet.row_values(i)) return li #脚本运行，首先执行它(一般将前置条件放在这里,类似init) def on_start(self): #用户名和密码 self.logindata=self.read_excel('./userDatas.xlsx') def on_stop(self): self.logout() # 括号里面参数表示该行为挑选执行的权重，数值越大，执行频率越高，不设置默认是1 # @seq_task(2)指定它的任务顺序执行，括号中的值为顺序值，数值越大，执行顺序越靠后 @task(1) def zentao_my(self): self.login() all_locusts_spawned.wait() r = self.client.get("/zentao/my/") self.locust_asser("我的地盘",r) #设置性能测试 class WebsiteUser(HttpLocust): #指向一个定义了的用户行为类 task_set = UserBehavior #用户执行任务之间等待时间的最小时间，单位毫秒 min_wait = 3000 #用户执行任务之间等待时间的最大时间，单位毫秒 max_wait = 6000 if __name__ == "__main__": os.system("locust - f demo3.py --host=http://192.168.x.xx:80 --no-web --csv=example -c 1 -r 1 -t 10s")
#多用户请求多个接口 import xlrd,os,random from locust import HttpLocust,TaskSet,task,events from gevent._semaphore import Semaphore #集合点 # 实例化 all_locusts_spawned = Semaphore() all_locusts_spawned.acquire() # 创建等待方法 def on_hatch_complete(**kwargs): all_locusts_spawned.release() # 当用户加载完成时触发 events.hatch_complete += on_hatch_complete class UserBehavior(TaskSet): #登录 def login(self): #随机取出列表中的用户名和密码 userInfo=self.randomValue(self.logindata) loginUrl ="/zentao/user-login.html/" h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0","Content-Type": "application/x-www-form-urlencoded"} body = {"account": userInfo[0],"password":userInfo[1],"keepLogin[]": "on","referer": "/zentao/my/"} r = self.client.post(loginUrl, data=body, headers=h) self.locust_asser("parent.location='/zentao/index.html'",r) # 退出 def logout(self): logoutUrl='http://192.168.x.xx:80/zentao/user-logout.html' h = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0", "Content-Type": "application/x-www-form-urlencoded"} r = self.client.get(logoutUrl,headers=h) self.locust_asser("/zentao/user-login.html", r) #断言 def locust_asser(self,expect,result): if expect in result.text: result.success() else: result.failure('Fail.') #取列表中的随机数 def randomValue(self,arr): r=random.randint(0,10000) index=r%len(arr) return arr[index] #读取excel，将每行数据放入一个列表，将所有列表放入一个列表形成二维列表，返回该二维列表 def read_excel(self,path): book = xlrd.open_workbook(path) sheet = book.sheets()[0] li = [] for i in range(1, sheet.nrows): li.append(sheet.row_values(i)) return li #脚本运行，首先执行它(一般将前置条件放在这里,类似init) def on_start(self): #用户名和密码 self.logindata=self.read_excel('./userDatas.xlsx') self.login() all_locusts_spawned.wait() def on_stop(self): self.logout() # 括号里面参数表示该行为挑选执行的权重，数值越大，执行频率越高，不设置默认是1 # @seq_task(2) 指定它的任务按顺序执行，括号中的数值为顺序值，数值越大，执行顺序越靠后 @task(1) class test(TaskSet): @task(2) def zentao_my(self): r = self.client.get("/zentao/my/") self.locust_asser("我的地盘",r) @task(1) def zentao_product(self): r = self.client.get("/zentao/product-browse.html/") self.locust_asser("需求列表",r) @task(1) def zentao_prject(self): r = self.client.get("/zentao/project/") self.locust_asser("项目首页",r) #设置性能测试 class WebsiteUser(HttpLocust): #指向一个定义了的用户行为类 task_set = UserBehavior #用户执行任务之间等待时间的最小时间，单位毫秒 min_wait = 3000 #用户执行任务之间等待时间的最大时间，单位毫秒 max_wait = 6000 if __name__ == "__main__": os.system("locust - f demo3.py --host=http://192.168.x.xx:80 --no-web --csv=example -c 1 -r 1 -t 10s")
2. windows有界面模式启动（代码中最后一行需要更改）
执行上面编写的脚本后，本地打开localhost:8089或127.0.0.1:8089即能正常访问locust的web UI界面，设置并发用户数，执行压测
Type：请求类型
Name：请求路径
requests：当前请求的数量
fails：当前请求失败的数量
Median：中间值，单位毫秒，一般服务器响应时间低于该值，而另一半高于该值
Average：所有请求的平均响应时间，毫秒
Min：请求的最小的服务器响应时间，毫秒
Max：请求的最大服务器响应时间，毫秒
Content Size：单个请求的大小，单位字节
reqs/sec：每秒钟请求的个数
3. windows命令行模式启动（代码中最后一行需要更改）
locust - f demo3.py --host=http://192.168.x.xx:80 --no-web --csv=example -c 1 -r 1 -t 10s
-c：指定要生成的Locust用户数
-r：每秒启动虚拟用户数
-t：如果要指定测试的运行时间,运行时间单位，如果不写单位默认是s,也可以指定小时h,分钟m
--csv=example：保存CSV文件到当前脚本目录example_distribution.csv、example_requests.csv
4. 分布式运行（代码中最后一行需要更改）
1. 单机主从模式（locust 中如需使用 master-slave 模式启动多个进程（使用多核处理器的能力），先启动 master，然后再逐一启动若干个 slave。其中 slave 的节点数要小于等于本机的处理器数）
1. 在cmd中先启动一个master节点，mater节点不执行任务
locust -f demo3.py --master
2. 开多个cmd窗口，启动多个slave节点，比如我开四个窗口，依次执行以下命令
locust -f demo3.py --slave
3. 在浏览器中输入：localhost:8089
2. 多机主从模式（选择其中一台电脑，启动master节点，在其它机器上启动从属Locust节点）
注：master 和每一台 slave 机器，在运行分布式测试时都必须要有 locust 的测试文件和运行环境
1. 在cmd中先启动一个master节点，mater节点不执行任务
locust -f demo3.py --master
2. 在其他电脑上cmd中运行以下命令
locust -f demo3.py --slave --master-host=192.168.x.xx
© 著作权归作者所有
举报
发表评论
0/200
