关键词:
文章目录
前言
locust是一款由python编写的压测工具,可以很好的和python进行结合使用,功能强大。
locust文档是英文版的,不利于阅读,本文从文档出发,模拟真实的压测需求与场景,带你领略locust的强大。
安装
locust安装方便,直接使用pip3进行安装
pip3 install locust
检查是否安装成功
locust -V
基本使用
需求1: 项目紧急,我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,需要模拟100个用户同时访问,每个用户访问url频率在1-5秒之间随机选择,压测10分钟。
对于这样的需求,如何编写locust压测脚本呢?下面是最简单的入门脚本。
from locust import HttpUser, task, between
class QuickstartUser(HttpUser):
wait_time = between(1, 5)
host = "https://www.baidu.com"
@task
def hello_world(self):
self.client.get("/hello")
假设该脚本文件名称为locust_test.py,可以使用以及下命令启动脚本
locust -f locust_test.py --headless -u 100 -r 100 -t 10m --html report.html
更多参数会在后文详细列举。
-f locust_test.py //代表执行哪一个压测脚本
--headless //代表无界面执行
-u 100 //模拟100个用户操作
-r 100 //每秒用户增长数
-t 10m //压测10分钟
--html report.html //html结果输出的文件路径名称
最后html测试结果结果如下
User类
每个locust脚本都需要继承User类,在User类定义了每个用户的压测行为。在上面例子中,QuickstartUser类继承HttpUser类,而HttpUser类继承User类,那么User类中究竟有哪些东西呢。
@task装饰器
既然locust定义了一个用户类,那么必然每个用户需要执行对应的任务,@task装饰器就定义了一个任务,函数就是任务执行的过程,其中装饰器后面的参数可以定义一个int代表任务的权重。
需求2: 接到了个压测需求,url地址为"https://www.baidu.com" ,get请求,需要模拟100个用户同时访问,每个用户访问循环访问"/hello"和"/world"这两个接口,任务之间没有间隔,并且用户选择这两个接口的比例为1:3,压测10分钟。
import os
from locust import HttpUser, task, between
class QuickstartUser(HttpUser):
# wait_time = between(1, 5)
host = "https://www.baidu.com"
@task
def hello_world(self):
self.client.get("/hello")
@task(3)
def hello_world(self):
self.client.get("/world")
if __name__ == '__main__':
os.system("locust -f locust_test.py --headless -u 100 -r 100 -t 10m --html report.html")
这边我把wait_time = between(1, 5)注释掉了。任务执行没有间隔,类似每个用户执行wile True操作不停给服务器发送请求。@task(3)定义了任务权重,脚本期望的情况是每执行4次请求,3次是“/world” 1次是"/hello"
任务间隔
返回需求1,如果把需求换成每个用户访问url频率固定2秒,这个怎么办呢?这边引出了任务间隔的概念,分为以下3类:
- 任务随机间隔wait_time = between(a, b)
- 任务固定间隔wait_time =constant(a)
- 任务无间隔
怎么理解间隔呢?其实可以这么理解,压测其实就是模拟用户给服务器发请求,任务无间隔代表每个用户不停的while 1给服务器发请求,任务固定间隔可以理解成while 1 sleep固定时间给服务器发请求。
需求3: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,需要模拟100个用户同时访问,每个用户访问url频率固定2秒发一次,压测10分钟。
from locust import HttpUser, task, constant
class QuickstartUser(HttpUser):
wait_time = constant(2)
host = "https://www.baidu.com"
@task
def hello_world(self):
self.client.get("/hello")
@tag装饰器
当我们需要选取其中一个任务或者几个任务进行压测时,@tag装饰器就发挥作用了。
class QuickstartUser(HttpUser):
# wait_time = between(1, 5)
host = "https://www.baidu.com"
@task
@tag("task1")
def hello_world(self):
self.client.get("/hello")
@task(3)
@tag("task2")
def hello_world2(self):
self.client.get("/world")
@task(3)
@tag("task3")
def hello_world3(self):
self.client.get("/world")
当前有3个任务,我们需要只选取task2和task3运行怎么办 --tags task2 task2 参数就可以解决这个问题。
locust -f locust_test.py --headless -u 100 -r 100 -t 10m --html report.html --tags task2 task2
前置与后置
用户可以声明on_start方法和/或on_stop方法。用户将调用on_start方法开始运行时on_stop方法。对于任务集on_start方法在模拟用户开始执行该任务集时调用,并且on_stop当模拟用户停止执行该任务集时调用(当interrupt()或者用户被杀死)
HttpUser类
HttpUser类继承自User类,所以User类中的属性HttpUser类都有,唯一存在不同的是HttpUser类自己实现了http的请求方法。
使用方法与requests库相同,获取requests对象可以使用self.client的方式获取。这边重点介绍一下self.client的断言方式。
需求4: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,post请求,返回结果为json格式“code”:0,“msg”:“ok”,其中code0代表成功,-1代表失败,需要模拟100个用户同时访问,每个用户访问url频率固定2秒发一次,压测10分钟。
import json
import os
from locust import HttpUser, task, tag, constant
class QuickstartUser(HttpUser):
wait_time = constant(2)
host = "https://www.baidu.com"
@task
@tag("task1")
def hello_world(self):
with self.client.post("/hello", catch_response=True) as response:
try:
status_code = response.status_code
text = json.loads(response.text)
if status_code != 200:
response.failure("status_code断言失败")
elif text["code"] != 0:
response.failure("code断言失败")
elif response.elapsed.total_seconds() > 0.5:
response.failure("Request took too long")
except Exception as e:
response.failure(str(e))
常用参数
具体见官方文档:http://docs.locust.io/en/stable/configuration.html
命令行配置
这边列出几个常用的参数
- -f locust_test.py //代表执行哪一个压测脚本
- –headless //代表无界面执行
- -u 100 //模拟100个用户操作
- -r 100 //每秒用户增长数
- -t 10m //压测10分钟
- –html report.html //html结果输出的文件路径名称
- -H https://www.baidu.com //定义访问的host
- –csv=CSV_PREFIX //将当前请求统计信息存储到CSV格式的文件中。设置此选项将生成三个文件:[CSV_PREFIX]_stats.CSV,[CSV_PREFIX]_stats_history.CSV和[CSV_PREFIX]_failures.CSV
- –only-summary //只打印摘要统计
- –print-stats //在控制台中打印统计信息
配置文件配置
# master.conf in current directory
locustfile = locust_files/my_locust_file.py
headless = true
master = true
expect-workers = 5
host = http://target-system
users = 100
spawn-rate = 10
run-time = 10m
locust --config=master.conf
常用压测场景实战
高用户高并发场景压测
需求5: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,需要模拟4000个用户同时访问,每个用户访问url不间断,压测10分钟。
HttpUser无法支持高并发压测,这时候需要将HttpUser换成FastHttpUser即可
from locust import FastHttpUser, task
class QuickstartUser(FastHttpUser):
host = "https://www.baidu.com"
@task
def hello_world(self):
self.client.get("/hello")
每个用户循环取数据
需求6: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,有个参数叫id,捞取线上数据进行真实场景压测,每个用户循环读取数据。
from locust import FastHttpUser, task
def prepare_data():
# 准备ids数据,模拟从文件中读取
return [i for i in range(100)]
class QuickstartUser(FastHttpUser):
host = "https://www.baidu.com"
data = prepare_data()
def on_start(self):
self.index = 0
@task
def hello_world(self):
id = self.data[self.index]
print(id)
self.client.get("hello", params="id": id)
self.index += 1
self.index = self.index % (len(self.data))
保证并发测试数据唯一性,不循环取数据
需求7: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,有个参数叫id,捞取线上数据进行真实场景压测,保证每个id只请求一次,不循环取数据。
这边使用到了python内置类Queue,他是线程安全的。
from queue import Queue
from locust import FastHttpUser, task
def prepare_data():
q = Queue()
# 准备ids数据,模拟从文件中读取
for i in range(100):
q.put(i)
return q
class QuickstartUser(FastHttpUser):
host = "https://www.baidu.com"
data_queue = prepare_data()
@task
def hello_world(self):
id = self.data_queue.get(timeout=3)
print(id)
self.client.get("hello", params="id": id)
if self.data_queue.empty():
print("结束压测")
exit()
保证并发测试数据唯一性,循环取数据
需求8: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,有个参数叫id,捞取线上数据进行真实场景压测,保证每个id只请求一次,循环取数据。
from queue import Queue
from locust import FastHttpUser, task
def prepare_data():
q = Queue()
# 准备ids数据,模拟从文件中读取
for i in range(100):
q.put(i)
return q
class QuickstartUser(FastHttpUser):
host = "https://www.baidu.com"
data_queue = prepare_data()
@task
def hello_world(self):
id = self.data_queue.get()
print(id)
self.client.get("hello", params="id": id)
self.data_queue.put_nowait(id)
这边使用data_queue.put_nowait(id),id用完之后再塞回队列中
梯度增压
需求9: 我接到了个压测需求,url地址为"https://www.baidu.com/hello" ,get请求,现在需要对其进行梯度增压,每10分钟增加20个用户,压测3小时,以窥探其性能瓶颈。
import math
import os
from locust import HttpUser, LoadTestShape, task
class QuickstartUser(HttpUser):
host = "https://www.baidu.com"
@task
def hello_world(self):
self.client.get("hello")
class StepLoadShaper(LoadTestShape):
'''
逐步加载实例
参数解析:
step_time -- 逐步加载时间
step_load -- 用户每一步增加的量
spawn_rate -- 用户在每一步的停止/启动
time_limit -- 时间限制
'''
setp_time = 600
setp_load = 20
spawn_rate = 20
time_limit = 60 * 60 * 3
def tick(self):
run_time = self.get_run_time()
if run_time > self.time_limit:
return None
current_step = math.floor(run_time / self.setp_time) + 1
return (current_step * self.setp_load, self.spawn_rate)
if __name__ == '__main__':
os.system("locust -f test4.py --headless -u 10 -r 10 --html report.html")
这边就需要使用到LoadTestShape这个内置类,类似于开了个线程,每秒都会调用tick方法,返回一个元组(当前用户数,增长率)
非http协议压测
有些情况下,我们压测并不是基于http协议的,比如websocket协议压测,jce协议的压测,这时候就需要重写User类的client,满足自己的压测需求。下面是重写client重要两点:
- self.client: locust协议入口实例,我们只要重写一个实例给client即可。
- 以下两个事件,用来收集报告信息,否则写好后执行你会发现收集不到性能数据
events.request_failure.fire()
events.request_success.fire()
需求10: 我接到了个压测需求,基于websocket协议接口进行压测。
import os
import time
import websocket
from locust import task, events, User
class WebSocketClient(object):
def __init__(self, host):
self.host = host
self.ws = websocket.WebSocket()
def connect(self, burl):
start_time = time.time()
try:
self.conn = self.ws.connect(url=burl)
except websocket.WebSocketTimeoutException as e:
total_time = int((time.time() - start_time) * 1000)
events.request_failure.fire(request_type="websockt", name='urlweb', response_time=total_time, exception=e)
else:
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(request_type="websockt", name='urlweb', response_time=total_time,
response_length=0)
return self.conn
def recv(self):
return self.ws.recv()
def send(self, msg):
self.ws.send(msg)
class WebSocketUser(User):
"""
A minimal Locust user class that provides an XmlRpcClient to its subclasses
"""
abstract = True # dont instantiate this as an actual user when running Locust
def __init__(self, environment):
super().__init__(environment)
self.client = WebSocketClient(self.host)
class QuickstartUser(WebSocketUser):
url = 'ws://localhost:8000/v1/ws/marketpair'
def on_start(self):
self.client.connect(self.url)
@task
def hello_world(self):
self.client.send("ok")
if __name__ == '__main__':
os.system("locust -f locust_test.py --headless -u 10 -r 10 -t 10s --html report.html")
最主要是events.request_failure.fire和events.request_success.fire这两个用来收集性能数据,如果不写报告收集不到性能数据。
参考
- 《locust官方文档》http://docs.locust.io/en/stable/writing-a-locustfile.html
- 《locust-基于python的性能测试工具(进化篇)》 https://km.woa.com/group/571/articles/show/411249
- 《locust压测rpc协议》https://www.cnblogs.com/yhleng/p/10031209.html
locust的使用
一、简介Locust是一款使用Python编写的压力测试工具,本篇总结会介绍在实际测试过程中遇到的问题https://www.locust.io/使用Locust的原因是因为可以模拟的用户数量可以通过添加硬件来增长,相对于jmeter来说配置更加方便。二、... 查看详情
性能测试工具jmeter和locust比较
前言ApacheJMeter和Locust都是是最受欢迎的性能测试工具。当你想做性能测试的时候,你会选择什么样的测试工具呢?是会选择jmeter?locust?今天,笔者将根据自己使用经验,针对jmeter、locust常用的性能测试工具进行简单介绍和对比... 查看详情
locust安装及其简单使用----基于python的性能测试工具(代码片段)
1、已安装python3.6,安装步骤略 pip安装:pipinstalllocust检查locust是否安装成功 locust--help 2、安装 pyzmq IfyouintendtorunLocustdistributedacrossmultipleprocesses/machines,werecommendyoutoalsoinstallpyzmq. 如果你打算... 查看详情
性能测试利器-locust框架解析
01认识Locust说起性能测试工具,大家肯定想到的都是Jmeter,是的,由于其简单易用、功能强大,已经变成主流的压测工具之一。当需要实现一些高级功能的时候,可以使用Java语言对Jmeter进行扩展。但是很多小... 查看详情
locust介绍
Locust介绍Anopensourceloadtestingtool.一个开源性能测试工具。defineuserbehaviourwithpythoncode,andswarmyoursystemwithmillionsofsimultaneoususers.使用Python代码来定义用户行为。用它可以模拟百万计的并发用户访问你的系统。 性能工具对比LoadRunner&nbs... 查看详情
locust性能测试安装(代码片段)
Locust简介Locust是一款易于使用的分布式用户负载测试工具。它用于对网站(或其他系统)进行负载测试,并确定系统可以处理多少并发用户。这个想法是,在测试期间,一群蝗虫(Locust)会攻击你的网站。您定义了每个蝗虫Locust... 查看详情
locust安装(代码片段)
转:http://www.testclass.net/locust/install/ Locust是基于Python语言的一个性能测试库,如果要想使用它来做性能测试必须要先安装Python。Python安装,参考 Locust安装方式一:通过pip命令安装>pipinstalllocustCollectinglocustDownloadinglocust-0.... 查看详情
性能测试工具locust和jmeter比较-及相关书籍下载
ApacheJMeter™和Locust都是是最受欢迎的性能测试工具。JMeter和Locust-简介JMeter是久经考验的性能框架之一,其第一个版本大约在20年前发布。它是用纯Java语言编写的。最初,JMeter开发用于执行Web和FTP应用程序的负载测试。但是,现... 查看详情
如何使用 Locust 进行 UI 性能测试?
】如何使用Locust进行UI性能测试?【英文标题】:HowtouseLocustforUIperformancetesting?【发布时间】:2019-12-0508:18:27【问题描述】:我想使用Locust进行UI性能测试。如何获取HTML元素(img、列表等)的加载时间?谢谢【问题讨论】:【参... 查看详情
使用locust测试mqtt协议
...码1.本次压测是使用本机安装的apache-activemq-5.15.9当作MQTT服务器 查看详情
性能工具之locust工具get与post请求(代码片段)
最近在学习locust性能工具,发现locust性能工具脚本需要python基础才能写脚本,但是对于性能测试人员来说python是基本功夫。 在locust中get脚本怎么写,为了方便直接在代码运行调试,采用关闭web模式,通过参考官... 查看详情
locust性能测试1-环境准备与基本使用(代码片段)
Locust简介Locust是一款易于使用的分布式用户负载测试工具。它用于对网站(或其他系统)进行负载测试,并确定系统可以处理多少并发用户。这个想法是,在测试期间,一群蝗虫(Locust)会攻击你的网站。您定义了每个蝗虫Locust... 查看详情
张同乐-从零开始,打造高效可靠的locust性能测试
...言欢迎来到Locust负载测试的世界!Locust是一款开源的负载测试工具,它可以模拟成千上万的用户同时访问你的应用程序,以测试其性能和稳定性。这个工具具有易于使用、可扩展和高度可定制化等特点,因此被广泛应用于各种类... 查看详情
locust--hellp
1.Locust简介 Locust是使用Python语言编写实现的开源性能测试工具,简洁、轻量、高效,并发机制基于gevent协程,可以实现单机模拟生成较高的并发压力。官网:https://locust.io/主要特点如下:1)使用普... 查看详情
不会吧,都2020年了,还有人不知道jmeter和locust的区别?(代码片段)
...他们之前的区别。ApacheJMeter和Locust都是是最受欢迎的性能测试工具。当你想做性能测试的时候,你会选择什么样的测试工具呢?是会选择jmeter?locust?今天将根据自己使用经验,针对jmeter、locust常用的性能测... 查看详情
不会吧,都2020年了,还有人不知道jmeter和locust的区别?(代码片段)
...他们之前的区别。ApacheJMeter和Locust都是是最受欢迎的性能测试工具。当你想做性能测试的时候,你会选择什么样的测试工具呢?是会选择jmeter?locust?今天将根据自己使用经验,针对jmeter、locust常用的性能测... 查看详情
性能工具之locust工具get与post请求(代码片段)
...近在学习Locust性能工具,发现Locust性能工具脚本需要python基础才能写脚本,但是对于性能测试人员来说python是基本功夫。脚本示例在Locust中get脚本怎么写,为了方便直接在代码运行调试,采用关闭web模式,通... 查看详情
性能测试工具loucst使用(最新版本)(代码片段)
locust简介:基于python编写,简单易于上手;支持分布式;脚本编写容易,web图形化界面操作容易,结果简洁,易读。官网:Locust-Amodernloadtestingframeworkhttps://locust.io/一、环境搭建:1.搭建python环境2.安装... 查看详情