简介
Locust是一个使用Python编写的可扩展、分布式的开源性能测试工具。
优点
- 相比于Jmeter、LoadRunner这种基于GUI的工具而言，Locust使用Python语言来描述测试场景使模拟用户行为变得更加灵活和简洁，除了Http(s)协议之外，Locust可以测试任意协议的系统，只需要实现Python调用对应协议的库进行请求即可(类似HttpLocust类)。
- Locust的并发机制采用协程的方式，相比于进程和线程减少了系统级资源调度，因此单机的产生的并发能力相比于LoadRunner、jmeter得到了大幅的提升。
安装
pip install locust
Locust脚本编写
import queue
from locust import HttpLocust, TaskSet, task
from locust.clients import HttpSession
from sign import ParserData
class UserBehavior(TaskSet):
parser_data = ParserData() # 解析接口传参类
_client = 20
version = 119
user_info = None
@staticmethod
def get_user_info(response):
r = response.json().get('content')
return {
'market_id': r.get('marketId'),
'token': r.get('token')
}
def on_start(self):
try:
user, password = self.locust.users.get_nowait()
except queue.Empty:
print('test data run out. test ended.')
exit(0)
client = HttpSession(base_url='http://login.xxxx.cn')
data = self.parser_data(loginName=user,
password=password,
client=self._client,
version=self.version)
response = client.post(url='/login', data=data)
self.user_info = self.get_user_info(response)
self.locust.users.put_nowait((user, password))
@task(2)
def index(self):
url = '/index'
data = self.parser_data(market_id=self.user_info['market_id'],
client=self._client,
version=self.version,
pnum='3')
headers = {'Authorization': 'Barer:' + self.user_info['token'],
'Accept': 'application/vnd.hs-api.v1+json'}
with self.client.post(url=url, data=data, headers=headers,
verify=False, catch_response=True) as response:
if response.status_code == 200:
response.success()
else:
response.failure('http error.')
@task(1)
def shop_car_list(self):
url = '/shopCar/list'
data = self.parser_data(market_id=self.user_info['market_id'],
ischaidan='1',
client=self._client,
version=self.version)
headers = {'Authorization': 'Bearer:' + self.user_info['token'],
'Accept': 'application/vnd.hs-api.v1+json'}
with self.client.post(name='ShopCar', url=url, data=data, headers=headers,
verify=False, catch_response=True) as response:
if response.status_code != 200 or "失败" in response.text:
response.failure('response error.')
else:
response.success()
class Stay(TaskSet):
index = 0
def on_start(self):
self.index += 1
@task
def get_error(self):
response = self.client.get('/1', name='error', allow_redirects=False,
verify=False, catch_response=True)
if response.status_code == 200:
response.success()
else:
response.failure('http error.')
@task
def logout(self):
self.interrupt()
class User(TaskSet):
tasks = {Stay: 1}
@task(1)
def user(self):
self.client.get('/', verify=False)
class WebsiteUser(HttpLocust):
task_set = UserBehavior
host = 'https://xxxx.api.xxxx.cn'
min_wait = 1000
max_wait = 3000
users = queue.Queue()
users.put_nowait(('user1', '1232'))
users.put_nowait(('user2', '1234'))
users.put_nowait(('user3', '1321'))
weight = 3
stop_timeout = 20
class WebsiteU(HttpLocust):
task_set = User
host = 'https://www.baidu.com'
min_wait = 0
max_wait = 0
weight = 1
stop_timeout = 60
简单解释下：
- UserBehavior和WebsiteUser两个类实现测试场景使用3个用户账号，每个用户会去先登录，然后分别去查看首页和进入购物车页面
- 首先导入了需要用到的类，HttpLocust类为Locust子类，模拟客户端的请求类，Taskset类为任务集类，task为任务装饰器。
Taskset
-
UserBehavior为Taskset子类，该类主要用来定义每个虚拟用户的操作行为
-
Taskset子类中可以定义一个on_start方法在正式开始测试前只执行一次，相当与初始化操作(这里说的只执行一次是每个虚拟用户都会去执行一次)，比如获取登录token等操作。
-
teskset类中的每个任务都需要用@task(weight=1)装饰器去装饰为一个任务，weight为执行的权重，如果不装饰，locust不会认为这是一个任务，UserBehavior类中@task(1)、@task(2)装饰器表示3个用户里面有2个用户去模拟执行index方法，有1个用户去执行shop_car_list方法
-
taskset类中self.client属性请求操作时传入catch_response参数，设置为True，可以标记响应结果为成功或失败，即使响应是成功的，也可以标记为失败，默认为Fasle
-
taskset类中self.client属性请求操作时有一个name参数，当设置了name的值时，最后的请求结果展示中name字段会显示这里定义的name值，相当与给这个方法起了一个别名.
-
taskset类中interrupt(reschedule=True)方法在顶层的taskset类(即被指定到Locust子类中的taskset)中不可用，reschedule为True时，从被嵌套的任务中出来立即执行新的任务，如果为False从被嵌套的任务中出来会等待min_time-max_time之间的随机时间，然后再执行新的任务，这个方法主要用来跳出嵌套的任务集
HttpLocust
-
WebsiteUser为HttpLocust子类，该类是用来模拟用户的类，定义了一些用户信息，及请求方式
-
HttpLocust子类中task_set属性用来指定模拟用户执行的操作，即Taskset子类
-
HttpLocust子类中的host属性为被测试系统的host，当命令行中没有指定--host参数时，此属性会生效
-
HttpLocust子类中的min_wait、max_weight为最大等待时间和最小等待时间，每个请求会从这两个时间间隔中随机取一个时间等待，相当用户实际操作系统时每个动作的思考时间。若测试单个接口则对应的值都设置为0即可。如果Taskset类中定义了min_wait、max_weight则会覆盖Locust子类中定义的值。单位ms，默认值1000ms
-
HttpLocust子类中的stop_timeout属性为执行测试的时间，单位为s
-
HttpLocust子类中的weight为该类执行的权重，当有多个子类时生效，如WebsiteUser、WebsiteU两个HttpLocust子类中weight值分别为3和1.
-
Locust默认单机单进程运行，此模式下并不能充分利用单机的多处理器，可使用分布式运行，即开启一个master，n个slave(n为处理器个数)，master负责启动Locust的web服务和任务分发，不会产生压力，slave主要负责产生压力
运行模式：
no-web模式
no_web模式指在命令行中直接运行
locust -f load_test.py -c 1 -r 1 -n 1
- -f 指定要运行的Locust性能测试文件
- -c 指定模拟的并发用户数
- -r 指定每秒的启动用户数
- -n 指定运行次数
- -t 指定运行的时间，例如300s,1m,1h
写完脚本调试时可在该模式下运行
单机单进程运行
locust -f load_test.py
分布式运行
分布式运行，有单机多进程运行和多机多进程运行两种
locust -f load_test.py --master master模式下启动locust
locust -f load_test.py --slave 启动一个locust slave节点，单机多进程模式
locust -f load_test.py --slave --master-host=192.168.105.11 启动一个locust slave节点，多机模式下
no_web模式下运行
image.png
web模式
image.png
- Number of users to simulate：需要模拟的虚拟用户个数
-
Hatch rate (users spawned/second)：启动虚拟用户的速率，每秒产生出多少个用户数
QQ图片20171109190334.png
- 显示并发数、响应时间、异常率、每秒请求数等
-
reqs/sec(每秒请求数)为根据最近2s请求数据计算得到的数据，即瞬时值
QQ图片20171109190430.png
-
显示rps、响应时间、并发数在整个测试运行中的走势图
QQ图片20171109190455.png
- 显示测试过程中出现的所有失败的请求
Exceptions显示测试过程中出现的异常
Download Date提供测试结果csv文件的下载
测试数据：
- locust子类中设置的数据是全局的，为list时，使用自增方式取数据，并发运行时会出现取到的数据重复的情况，如果对数据唯一性有要求，使用python的queue队列的数据结构即可
- taskset子类中设置的数据是局部的，即每一个虚拟用户都会有一个属于自己的这个变量
queue数据结构
队列形式
q=queue.Queue(maxsize=3) 先进先出队列
q.put(1) 向队列中存数据
q.get() 向队列中取数据
q.put_nowait() 相当于q.put(1, block=False)，当q队列满了之后put会触发queue.Full异常
q.get_nowait() 相当与q.get(block=False),当q队列为空之后get会触发queue.Empty异常
网友评论