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 os
import locust
from locust import task, TaskSet, HttpUser, constant
# 任务集 用户行为脚本
class UserBehavior(TaskSet): # 必须继承TaskSet
wait_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 csv
import os, requests
import queue
from locust import TaskSet, task, HttpUser
from 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 s
class 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")
