-
安装 locust
pip install locust
- 查看是否安装成功，查看版本号 locust -V
- 更新最新版本号 pip install -U --pre locust
demo：
from locust import FastHttpUser, TaskSet, between, task, events, LoadTestShape
import os
from queue import Queue
from gevent._semaphore import Semaphore
all_locusts_spawned = Semaphore()
all_locusts_spawned.acquire()
n = 0
def on_hatch_complete(**kwargs):
"""Select_task类的钩子方法"""
all_locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
class UserBehavior1(TaskSet):
def login(self):
global n
n += 1
# print("%s个虚拟用户开始启动，并登录"%n)
def logout(self):
print("退出登录")
def on_start(self):
self.login()
self.id = self.parent.q.get()
self.token = self.parent.test_data[self.id%4]
# all_locusts_spawned.wait() # 集合点
print(f"{self.id}任务进入on_start")
def on_stop(self):
print(f"{self.id}任务退出on_stop")
self.parent.q.put(self.id)
@task
def test_baidu_localnews(self):
url = '/widget?id=LocalNews&ajax=json&t=1632650885696'
param = {"limit": 2,"offset": 0,}
# print(f"--------{self.token}------")
with self.client.get(url,params=param,headers={},catch_response = True) as response:
if response.json().get('errno') == 0:
response.success()
else:
response.failure(f'{self.id}------Failed!')
class UserBehavior2(TaskSet):
@task(5)
def test_baidu_error(self):
url = '/widget?id=China&ajax=json&t=1632650885696'
param = {"limit": 2,"offset": 0,}
with self.client.get(url, params=param, headers={}, catch_response=True) as response:
if response.json().get('errno') == 0:
response.success()
else:
response.failure(f'Failed!')
@task(5)
def test_baidu_ad(self):
url = '/widget?id=ad&ajax=json'
param = {"limit": 2, "offset": 0, }
with self.client.get(url, params=param, headers={}, catch_response=True) as response:
if response.json().get('errno') == 0: #.status_code == 200:
response.success()
else:
response.failure('Failed!')
# class MyCustomShape(LoadTestShape): # 自定义并发数
# stages = [
# {"time": 10, "users": 10, "spawn_rate": 10},
# {"time": 20, "users": 20, "spawn_rate": 10},
# {"time": 30, "users": 10, "spawn_rate": 10},
# {"time": 40, "users": 20, "spawn_rate": 10},
# ]
# def tick(self):
# run_time = self.get_run_time()
# for stage in self.stages:
# if run_time < stage['time']:
# tick_data = (stage['users'],stage['spawn_rate'])
# return tick_data
# return None
class WebsiteUser(FastHttpUser):
q = Queue()
for i in range(1, 1000):
q.put(i)
host = 'http://news.baidu.com'
# tasks = {UserBehavior1: 1, UserBehavior2: 2}
tasks = [UserBehavior1]
test_data = ['1300000000', '1300000001', '1300000002', '1300000003']
wait_time = between(0.9, 1.1)
if __name__ == '__main__':
os.system("locust -f main.py")
脚本解析：
1、通过@task(n)装饰的方法为一个事务，方法的参数用于指定该行为的执行权重，参数越大每次被虚拟用户执行的概率越高，默认为1
2、 wait_time = between(1, 3)
来模拟用户的真实操作，例如用户在实际操作中，从上一步到下一步的操作，中间可能存在思考时间 ，between(1, 3) 表示用户思考的时间，随机为1到3秒
3、 host = 'http://news.baidu.com' 为要测试的域名
4、on_start ：每个虚拟用户在启动时都会调用该方法，获取token或测试用户数据等都放在这里
on_stop ： 当虚拟用户用户停止运行（被终止）时调用
每个虚拟用户只执行一次on_start和on_stop函数
5、HttpUser只被执行一次，因此很多公共的东西可以写到这里，所带属性在taskset中可以通过parent取出来。
6、 Semaphore常用于设置集合点（等待所有user都创建，再运行），用于对某些接口进行狭义的压力测试，以便进行调优。通常不用集合点，也就是这部分代码可以删除
7、检查点，请求函数需设定catch_response=True，会根据success failure进行统计
8、LoadTestShape可以自定义并发数量和时间，来模拟实际场景：时间峰值策略、时间阶段负载策略、逐步负载策略等等
locust性能测试：（十）自定义负载策略 - 简书
参数解析： Type：请求的类型，例如GET/POST。 Name：请求名称 requests：当前已完成的请求数量 fails：当前失败的数量 Median：响应时间的中间值，单位为毫秒 90%ile：根据正态分布，90%的响应时间在正态分布平均值下方 Average：平均响应时间，单位为毫秒 Min：最小响应时间，单位为毫秒 Max：最大响应时间，单位为毫秒 average Size：平均每个请求的数据量，单位为字节 current RPS（requests per second）：每秒钟处理请求的数量，即RPS
