locust为python的一个库，pip install locust直接安装。
locust摒弃了进程和线程，采用协程（gevent）的机制，单台测试机可以产生数千并发压力。jmeter为线程，所以单台负载机并发数比不上locust。
直接上代码
新建py文件 locustfile.py
on_start 相当于setup，task执行前执行一次，可以理解为数据初始化
from locust import HttpUser, TaskSet, task
import requests,hashlib,json,base64
class WebsiteTasks(TaskSet):
def on_start(self):
print("start")
url ="http://www.xxx.com"
timestamp_res = requests.get(url)
self.timestamp = json.loads(timestamp_res.text)['data']
self.bizCode ="SO1234567898765"
self.publicKey ="sdfsdfsdfsdfsdf"
self.secretKey ="sdfsdfsdfsdfsdf"
self.md5_query = hashlib.md5(b'{}{}{}').hexdigest().format(self.bizCode,self.secretKey,self.timestamp)
self.digest=str(base64.b64encode(self.md5_query.encode(encoding='utf-8')))[2:-1]
self.headers = {"Content-Type":"application/json"}
print(self.md5_query,self.digest)
catch_response=True可以理解为该请求允许被标记为失败，也就是断言
@task(1)
def order_query(self):
url ="/api/xxx/xxx/xxx/xxx/xxx"
data = {"bizCode":self.bizCode,"digest":self.digest,"publicKey":self.publicKey,"timestamp":self.timestamp}
with self.client.post(url=url,data=json.dumps(data),headers=self.headers,catch_response=True)as response:
#断言状态码
if response.status_code ==200:
response.success()
#断言文本
if json.loads(response.text)["data"]["xxx"] =="xxxxx":
response.success()
else:
response.failure("failed")
class WebsiteUser(HttpUser):
tasks = [WebsiteTasks]
host ="http://www.xxxxx.cn"
min_wait =1000
max_wait =5000
# if __name__ == "__main__":
# import os
# os.system("locust -H http://www.xxxxxx.cn -f locustfile.py")
在命令行中运行locust -H http://www.xxxxxx.cn -f locustfile.py。
就可以在浏览器中输入localhost:8089打开locust的web界面.
number of total users to simulate代表总的用户数
spawn rate 代表每秒增加多少。
输入完成后 直接start swarming就开始了性能测试。
测试界面有报告，性能图等
