Locust basiert auf dem Coroutine-Mechanismus von Python, der die Einschränkungen von Thread-Prozessen durchbricht und eine hohe Parallelität auf einer Testmaschine ausführen kann.
Gleichzeitig an den Server senden. Der Zweck besteht darin, herauszufinden, ob im System Transaktionskonflikte oder Sperraktualisierungen vorliegen Wenn es Probleme mit der Installation gibt, können Sie es über die Douban-Quelle herunterladen kann Locust direkt im Browser öffnen
Rendezvous-Punkt lr_rendezvous
Natürlich können wir den Rendezvous-Punkt-Vorgang ausführen. Geben Sie das Setup der obigen Vorlage ein, um es auszuführenpip install locust
In den beiden oben genannten Beispielen haben wir einige gesehen , wie z. B. der Dekorator events.test_start.add_listener; events.test_stop.add_listener wird vor und nach dem Auslastungstest verwendet. Führen Sie einige Vorgänge wie on_start und on_stop aus, z. B. kann die Aufgabe zum Zuweisen verwendet werden Gewicht der Aufgabe
Wartezeitfrom 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
Sie können wait_time auch in der User-Klasse umschreiben, um eine automatische Definition von
Tag-Markierunglocusts_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"')
Durch Markieren von Aufgaben können Sie bestimmte Aufgaben zur Laufzeit ausführen: # 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)
@tag('test1') @task(2) def index(self): self.client.get('/yetangjian/p/17320268.html')
Definieren Sie selbst eine shape.py, indem Sie LoadTestShape erben und das Häkchen überschreiben.
Diese Formklasse erhöht die Anzahl der Benutzer in Einheiten von 100 Blöcken mit einer Rate von 20 , und stoppen Sie dann den Belastungstest nach 10 Minuten (ab der 51. Sekunde des Laufs wird user_count auf 100 gerundet). über die Befehlszeile
#只执行标记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"')
Das obige ist der detaillierte Inhalt vonSo installieren und verwenden Sie das leichte Python-Leistungstool Locust. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!