1.locust特点:

1.1 支持Python编写测试用例方案;

1.2 使用requests发送http请求;

1.3 使用协程实现,高并发时消耗更低;

1.4 使用Flask提供 Web UI;

1.5 有第三方插件支持扩展;

2.创建locust 性能测试用例步骤:

2.1 创建locust.TaskSet的子类;

2.2 为用例加上@locust.task()装饰器;

2.3 使用self.client发送请求;

2.4 制定await_time属性;

import osimport locustfrom locust import task, TaskSet, HttpUser, constant# 任务集 用户行为脚本class UserBehavior(TaskSet):# 必须继承TaskSetwait_time = locust.between(1, 2)# 每个task间隔1~2秒@locust.task(1)#@task装饰该方法表示为用户行为,括号里面参数表示该行为的执行权重:数值越大,执行频率越高,不设置默认是1;def getBaidu(self):# headers根据自己的接口实际情况填写即可headers = {'content-type': 'application/x-www-form-urlencoded', 'ua': 'xxx', 'Cookie': 'xxx', }url = "https://www.baidu.com/"res = self.client.get(url, headers=headers)print(res.status_code)assert res.status_code == 200# wait_time = constant(1)# 每个用户在每次任务执行之间等待5秒class WebsiteUser(HttpUser):#WebsiteUser()类用于设置生成负载的基本属性:tasks = [UserBehavior]#task_set:指向定义了用户行为的类min_wait = 500#min_wait:模拟负载的任务之间执行时的最小等待时间,单位为毫秒max_wait = 1000#min_wait:模拟负载的任务之间执行时的最大等待时间,单位为毫秒host = "http://localhost:8089"if __name__ == "__main__":os.system("locust -f %s " % __file__)# 此处导入os,可以在pycharm中直接运行此py文件

3.locust执行压测

3.1通过web UI执行

locust -f {用例文件.py}

3.2 通过命令行执行

locust -f {文件名.py} –headless -u 500 -r 10 -t 1h30m [–host 192.168xx]

-f:指定运行的用例文件;
–headless:非GUI方式执行;
-u:指定多少并发用户数;
-r:指定每秒启动多少用户;
-t:指定压测运行时间;
–host:指定base host参数;

4.数据关联:

通过变量self.token=res.json()[“token”]方式,token可被其他方法使用。

5.参数化:

引入队列的概念 queue ,实现方式是将参数推入队列,测试时依次取出,全部取完后 locust 会自动停止。若是使用参数循环压测,需要将取出的参数再推入队尾。

import csvimport os, requestsimport queuefrom locust import TaskSet, task, HttpUserfrom requests.packages.urllib3.exceptions import InsecureRequestWarning# 禁用安全请求警告requests.packages.urllib3.disable_warnings(InsecureRequestWarning)def fnReadData():f = open("uuid.text", "r") #读取参数文件data = []#声明空列表data = csv.reader(f)#通过 csv读取文件内容s = queue.Queue() #实例化一个queue对象for each in data: #循环读取open里面的数据for key in each:try:s.put_nowait(key)#put到队列中except queue.Full:print("Queue overflow")f.close()return sclass MyBlogs(TaskSet):# 访问我的博客首页@task(1)def get_blog(self):# 定义请求头header = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36"}data = self.user.queueData.get()req = self.client.get("/357712148/%s" % data, headers=header, verify=False)if req.status_code == 200:print("success")else:print("fails")class httpGet(HttpUser):tasks = [MyBlogs]min_wait = 3000# 单位为毫秒max_wait = 6000# 单位为毫秒queueData = fnReadData()# 队列实例化if __name__ == "__main__":#通过好os.system("locust -f lcome.py --host=https://blog.51cto.com--headless -u 1 -r 1 -t 1s")