locust简易强大的性能测试工具
posted on 2018-12-30 by wangyang thanks 我来翻,locust帮助文档
locust是python编写的压力测试工具,基于python的request库发送并发请求,安装简单强大,对熟悉python的玩家来说,非常容易上手locust官网
安装就一句话,推荐使用2.7.12以上版本python
pip install locust
快速上手下面的例子可以跑起来一个最基础的测试脚本
创建一个locusttest.py的文件
from locust import HttpLocust, TaskSet, task
class UserBehavior(TaskSet):
@task(2) #task作为装饰器,括号里写数字表示了迭代的权重
def index(self):
self.client.get("/")
@task(1)
def index2(self):
self.client.get("/blog/locust")
class WebsiteUser(HttpLocust):
host = "http://blog.jcgame.net"
task_set = UserBehavior
min_wait = 5000 #两次最小时间间隔
max_wait = 9000
写完保存好,然后cd到脚本目录下,locust -f locusttest.py
此时会启动server,用浏览器打开localhost:8089 即可如图

小技巧:
在做性能测试中,经常会有一些特别的需求,如有如下流程需求 1.客户端登陆之后,服务器返回token 2.随后所有的逻辑请求,需要携带服务器返回token才可发送 3.登陆采用的id是唯一的,每个虚拟用户只能固定使用1个id 4.登陆使用的是json传输 5.主从启动:locust -f locust.py --master --port=8088 & 6.locust -f locust.py --slave --master-host=172.0.0.1:8088 & 下面的例子实现了上面的1234,on_start里调用login,保证每个用户只执行一次
from locust import HttpLocust, TaskSet, task
headers={"Content-type": "application/json"}
class UserBehavior(TaskSet):
def guest_login(self, deviceId):
deviceId = deviceId
checksum = sha1.customer_sha1(deviceId) #自定义加密
data = {
"deviceId": deviceId,
"checksum": checksum,
"platform": 3,
"adid": "",
"name":{"id":1}
}
data = json.dumps(data)
response = self.client.post("/login", data, headers=headers, verify=False)
if response.status_code == 200:
self.content = json.loads(response.content)
else:
response.failure(response)
def on_start(self): #locust提供了多种初始化方法 单个用户on_start只会被执行一次,其他方法请参考官方文档
#https://docs.locust.io/en/stable/
deviceId = self.locust.user_data_queue.get()
self.guest_login(deviceId)
@task
def update_profile(self):
profile = json.dumps(get_local_profile("test.json")) # str to dict
if self.content["errno"] == 0:
secret = self.content['data']['secret']
token = self.content['data']['token']
checksum = cooking_sha1.cooking_sha1(profile, secret)
response = self.client.post("/update?token=%s&checksum=%s" % (token, checksum), profile, headers=headers, verify=False)
#一定要加header,否则多重dict无法被服务端正确解析
check_response(response)
else:
print self.content
class WebsiteUser(HttpLocust):
task_set = UserBehavior
host = "https://blog.jcgame.net"
min_wait = 500
max_wait = 2000
user_data_queue = Queue.Queue() #使用队列方式保证id唯一
with open("test_deviceId.txt") as r: # get login id
for line in r.readlines():
user_data_queue.put_nowait(line.strip("\n"))
常见问题
问题1:locust和其他性能测试工具比优势在哪
解决:locust使用协程方式进行加压,单机拥有更高的并发量
问题2:linux可以部署么?
解决:官方文档推荐使用linux
问题3:上手有多快
解决:复制上面的例子直接可以用

协程讲解:什么是协程