Rumah > Artikel > pembangunan bahagian belakang > Cara memasang dan menggunakan alat prestasi ringan Python Locust
Locust adalah berdasarkan mekanisme coroutine python, yang memecahkan had proses benang dan boleh menjalankan konkurensi tinggi pada mesin ujian
1. Kelajuan: ukur sistem prestasi Kecekapan pemprosesan: masa tindak balas
2. Berapa banyak: mengukur kapasiti pemprosesan sistem: berapa banyak transaksi (tps) boleh diproses setiap unit masa
Ujian prestasi adalah berdasarkan pada titik keperluan ujian yang paling biasa Untuk tiga kategori berikut
1 Muatkan ujian beban ujian
Sentiasa menekan pelayan, yang layak untuk penunjuk yang telah ditetapkan atau beberapa sumber sistem yang mencapai kesesakan, tujuannya adalah untuk mencari kapasiti beban maksimum sistem
2 Ujian tekanan
Sahkan sama ada sistem stabil dengan mengekalkan beban tinggi untuk masa yang lama
3 Ujian konkurensi:
Serahkan permintaan kepada pelayan pada masa yang sama, penemuan tujuan Sama ada terdapat konflik transaksi atau peningkatan kunci dalam sistem
Model beban prestasi
pemasangan wujud Soalan, anda boleh memuat turun
pip install locust
Pada asasnya, dalam kebanyakan senario kita boleh buat pengubahsuaian berdasarkan templat ini 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 = 5000
Nota: Di sini kami Hos web diberikan supaya kami boleh membuka belalang terus dalam pelayar
Sudah tentu kita boleh meletakkan operasi pertemuan ke dalam persediaan templat di atas untuk menjalankannya
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"')
Kami telah melihat beberapa dalam dua contoh di atas, seperti acara penghias.test_start.add_listener ; events.test_stop.add_listener digunakan untuk melaksanakan beberapa operasi sebelum dan selepas ujian bebanan. Contoh lain ialah on_start, on_stop, yang dijalankan sebelum dan selepas pelaksanaan tugasan
dan masa menunggu
# 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)
juga boleh digunakan dalam kelas Pengguna Hantar dan tulis semula masa_tunggu untuk mencapai penyesuaian
tag tag
@tag('test1') @task(2) def index(self): self.client.get('/yetangjian/p/17320268.html')
Dengan menandakan tugasan, anda boleh melaksanakan tugas tertentu pada masa jalan:
#只执行标记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"')
Penyesuaian gagal
#定义响应时间超过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,但我定义为失败")
Bentuk beban tersuai
Sesuaikan bentuk. kadar 20, dan kemudian hentikan ujian beban selepas 10 minit (kiraan_pengguna akan meningkat daripada saat ke-51 selepas pusingan larian kepada 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
Carta operasi adalah seperti berikut
Pencetus
os.system('locust -f read.py,shape.py --web-host="127.0.0.1"')pada peringkat masa yang berbeza melalui baris arahan Contoh
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 None
Atas ialah kandungan terperinci Cara memasang dan menggunakan alat prestasi ringan Python Locust. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!