Locust is based on python's coroutine mechanism, which breaks the limitations of thread processes and can run high concurrency on a test machine
1. Speed: measure the performance of the system Processing efficiency: response time
2. How much: measures the processing capacity of the system: how many transactions (tps) can be processed per unit time
Performance testing is based on the most common points of test requirements For the following three categories
1 Load testing load testing
Continuously pressurize the server, worthy of predetermined indicators or some system resources reaching bottlenecks, the purpose is to find the maximum load capacity of the system
2 Stress test
Verify whether the system is stable through high load for a long time
3 Concurrency test:
Submit requests to the server at the same time, and the purpose is discovered Whether there is transaction conflict or lock escalation in the system
Performance load model
Installation exists Question, you can download the locust template through Douban source
pip install locust
Basically in most scenarios we can make modifications based on this template read.py
from locust import HttpUser, TaskSet, task, tag, events
# 启动locust时运行
@events.test_start.add_listener
def setup(environment, **kwargs):
# print("task setup")
# 停止locust时运行
@events.test_stop.add_listener
def teardown(environment, **kwargs):
print("task teardown")
class UserBehavor(TaskSet):
#虚拟用户启用task运行
def on_start(self):
print("start")
locusts_spawned.wait()
#虚拟用户结束task运行
def on_stop(self):
print("stop")
@tag('test1')
@task(2)
def index(self):
self.client.get('/yetangjian/p/17320268.html')
@task(1)
def info(self):
self.client.get("/yetangjian/p/17253215.html")
class WebsiteUser(HttpUser):
def setup(self):
print("locust setup")
def teardown(self):
print("locust teardown")
host = "https://www.cnblogs.com"
task_set = task(UserBehavor)
min_wait = 3000
max_wait = 5000Note: Here we A webhost is given so that we can open locust directly in the browser
Of course we can put the rendezvous operation into the setup of the above template to run it
locusts_spawned = Semaphore()
locusts_spawned.acquire()
def on_hatch_complete(**kwargs):
"""
select_task类的钩子函数
:param kwargs:
:return:
"""
locusts_spawned.release()
events.spawning_complete.add_listener(on_hatch_complete)
n = 0
class UserBehavor(TaskSet):
def login(self):
global n
n += 1
print(f"第{n}个用户登陆")
def on_start(self):
self.login()
locusts_spawned.wait()
@task
def test1(self):
#catch_response获取返回
with self.client.get("/yetangjian/p/17253215.html",catch_response=True):
print("查询结束")
class WebsiteUser(HttpUser):
host = "https://www.cnblogs.com"
task_set = task(UserBehavor)
wait_time = between(1,3)
if __name__ == '__main__':
os.system('locust -f read.py --web-host="127.0.0.1"')We have seen some in the above two examples, such as the decorator events.test_start.add_listener; events.test_stop.add_listener is used to perform some operations before and after the load test. Another example is on_start, on_stop, which runs before and after task execution. Another example is task, which can be used to allocate the weight of the task.
Waiting time
# wait between 3.0 and 10.5 seconds after each task #wait_time = between(3.0, 10.5) #固定时间等待 # wait_time = constant(3) #确保每秒运行多少次 constant_throughput（task_runs_per_second） #确保每多少秒运行一次 constant_pacing(wait_time)
can also be used in the User class Issue rewritten wait_time to achieve custom
tag tag
@tag('test1')
@task(2)
def index(self):
self.client.get('/yetangjian/p/17320268.html')By tagging tasks, you can execute certain tasks at runtime:
#只执行标记test1 os.system('locust -f read.py --tags test1 --web-host="127.0.0.1"') #不执行标记过的 os.system('locust -f read.py --exclude-tags --web-host="127.0.0.1"') #除去test1执行所有 os.system('locust -f read.py --exclude-tags test1 --web-host="127.0.0.1"')
Customization failed
#定义响应时间超过0.1就为失败
with self.client.get("/yetangjian/p/17253215.html", catch_response=True) as response:
if response.elapsed.total_seconds() > 0.1:
response.failure("Request took too long")
#定义响应码是200就为失败
with self.client.get("/yetangjian/p/17320268.html", catch_response=True) as response:
if response.status_code == 200:
response.failure("响应码200，但我定义为失败")
## Custom load shape
Customize a shape. py inherits LoadTestShape and overrides tick. This shape class will increase the number of users in units of 100 blocks at a rate of 20, and then stop the load test after 10 minutes (user_count will increase from the 51st second after running round to 100)from locust import LoadTestShape
class MyCustomShape(LoadTestShape):
time_limit = 600
spawn_rate = 20
def tick(self):
run_time = self.get_run_time()
if run_time < self.time_limit:
# User count rounded to nearest hundred.
user_count = round(run_time, -2)
return (user_count, self.spawn_rate)
return None
os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')
at different time stages Example
from locust import LoadTestShape
class StagesShapeWithCustomUsers(LoadTestShape):
stages = [
{"duration": 10, "users": 10, "spawn_rate": 10},
{"duration": 30, "users": 50, "spawn_rate": 10},
{"duration": 60, "users": 100, "spawn_rate": 10},
{"duration": 120, "users": 100, "spawn_rate": 10}]
def tick(self):
run_time = self.get_run_time()
for stage in self.stages:
if run_time < stage["duration"]:
tick_data = (stage["users"], stage["spawn_rate"])
return tick_data
return NoneThe above is the detailed content of How to install and use Python lightweight performance tool Locust. For more information, please follow other related articles on the PHP Chinese website!
![[Web front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
