locust壓力測試

安裝

pip install locust

驗證是否安裝成功

locust -V

使用

網上的教程基本上是前幾年的，locust已經更新了好幾個版本，有點過時了，在此做一個總結

啟動

默認是使用瀏覽器進行設置的

# 使用瀏覽器
locust -f .\main.py

其他參數

Usage: locust [options] [UserClass ...]常用選項:-h, --help            顯示幫助消息并退出-f <filename>, --locustfile <filename>包含測試的Python文件或模塊，例如“my_test.py”。接受多個逗號分隔.py文件、包名/目錄或指向的url遠程locustfile。默認為“locustfile”。--config <filename>   從中讀取其他配置的文件。-H <base url>, --host <base url>主機負載測試，格式如下：https://www.example.com-u <int>, --users <int>并發Locust用戶的峰值數量。主要使用與無頭或自動啟動一起。可以是在測試期間通過鍵盤輸入w、W（spawn1，10個用戶）和s，S（停止1，10用戶）-r <float>, --spawn-rate <float>生成用戶的速率為（每秒用戶數）。主要是與--headless或--autostart一起使用-t <time string>, --run-time <time string>在指定的時間量后停止，例如（300s，20m、3h、1h30m等）。僅與一起使用--headless 或 --autostart。默認為永遠運行。-l, --list            顯示可能的用戶類列表并退出--config-users [CONFIG_USERS ...]用戶配置為JSON字符串或文件。列表參數或JSON配置數組Web UI 選項:--web-host <ip>       將web界面綁定到的主機。默認為“*”（所有接口）--web-port <port number>, -P <port number>運行web主機的端口--headless            禁用web界面，然后立即開始測試。使用-u和-t來控制用戶數量和運行時間--autostart           立即開始測試（比如--headless，但是不禁用web UI）--autoquit <seconds>  完全退出Locust，運行后X秒完成。僅與 --autostart一起使用。這個默認設置是保持Locust運行，直到您按下CTRL+C關閉它--web-login           使用登錄頁面保護web界面。--tls-cert <filename>用于提供服務的TLS證書的可選路徑HTTPS--tls-key <filename>  用于提供服務的TLS私鑰的可選路徑HTTPS--class-picker        在web界面中啟用選擇框進行選擇從所有可用的用戶類和形狀類主選項：
在分布式運行Locust時運行LocustMaster節點的選項。主節點需要連接到它的Worker節點，然后才能運行負載測試。--master              將locust作為主節點啟動，工作節點連接到該節點。--master-bind-host <ip>主機監聽的IP地址，例如'192.168.1.1'. 默認為*（全部可用接口）。--master-bind-port <port number>主機監聽的端口。默認為5557。--expect-workers <int>延遲開始測試，直到達到此數量的workers（僅與結合使用 --headless/ --autostart）。--expect-workers-max-wait <int>master等待workers連接時間。默認為永遠等待--enable-rebalancing  如果在測試運行期間添加或刪除了新的worker，請重新分配用戶。實驗。Worker選項：
在分布式運行Locust時運行LocustWorker節點的選項。
通常只需要在workers上指定這些選項（和--locostfile），因為其他選項（-u、-r、-t、…）由主節點控制。--worker              將locust設置為以分布式模式運行，并將此進程設置為worker。可以與設置相結合--locustfile設置為“-”，以便從master下載。--processes <int>     分流 locust 進程的數量，以啟用系統。結合--worker標志或讓它自動設置--worker和--master標志，以實現一體化解決方案。在 Windows 上不可用。實驗。--master-host <hostname>要連接的locust主節點的主機名。默認值127.0.0.1。--master-port <port number>主節點上要連接的端口。默認值為5557。標記選項：
Locust任務可以使用@tag裝飾器進行標記。這些選項允許指定在測試期間包含或排除哪些任務。-T [<tag> ...], --tags [<tag> ...]測試中要包含的標簽列表，只會執行至少有一個匹配標簽的任務-E [<tag> ...], --exclude-tags [<tag> ...]要從測試中排除的標簽列表，只會執行沒有匹配標簽的任務請求統計選項：--csv <filename>      將請求統計數據以CSV格式存儲到文件中。設置此選項將生成三個文件：<filename>_stats.csv、<filename>_stats_history.csv和<filename>_failures.csv。前綴文件夾部分將自動創建--csv-full-history    將每個統計條目以CSV格式存儲到_statshistory.csv文件。您還必須指定“--csv”參數來啟用此功能。--print-stats         在UI運行中啟用請求統計數據的定期打印--only-summary        禁用請求統計信息的定期打印--headless 運行--reset-stats         生成完成后重置統計信息。在分布式模式下運行時，應同時在master和worker上設置--html <filename>     將HTML報告存儲到指定的文件路徑--json                將JSON格式的最終統計數據打印到stdout。可用于解析其他程序/腳本中的結果。與--headless和--skip-log一起使用，僅用于json數據的輸出。日志記錄選項：--skip-log-setup      禁用Locust的日志記錄設置。相反，配置是由Locust測試或Python默認值提供的。--loglevel <level>, -L <level>在DEBUG/INFO/WARNING/ERROR/CRITICAL之間進行選擇。默認值為INFO。--logfile <filename>  日志文件的路徑。如果未設置，日志將轉到stderr其他選項：--show-task-ratio     打印用戶類任務執行率表。如果某些類定義了非零的fixed_count屬性，請將其與非零--user選項一起使用。--show-task-ratio-json打印User類任務執行率的json數據。如果某些類定義了非零的fixed_count屬性，請將其與非零--user選項一起使用。--version, -V         顯示程序的版本號并退出--exit-code-on-error <int>設置測試結果包含任何失敗或錯誤時使用的進程退出代碼。默認為1。-s <number>, --stop-timeout <number>退出前等待模擬用戶完成所有執行任務的秒數。默認情況是立即終止。當運行分布式時，只需要在主服務器上指定。--equal-weights       使用均勻分布的任務權重，覆蓋locostfile中指定的權重。User classes:<UserClass1 UserClass2>在命令行末尾，您可以列出要使用的User類（可用的User類可以用--list列出）。LOCUST_USER_CLASS環境變量也可用于指定USER類。默認設置是使用所有可用的用戶類示例：locust -f my_test.py -H https://www.example.comlocust --headless -u 100 -t 20m --processes 4 MyHttpUser AnotherUser有關更多詳細信息，包括如何使用文件或環境變量設置選項，請參閱文檔：
https://docs.locust.io/en/stable/configuration.html

參數介紹在其他文章上有有很多，這里不多贅述，下面主要介紹一下項目中的應用

`TaskSet`

新版的locust使用的方式與舊版有點區別，寫用例時需要繼承 TaskSet來實現

代碼如下：

# coding: utf-8
from locust import TaskSet
from locust.clients import ResponseContextManagerfrom .settings import USER_AGENTclass UserBehaviorBase(TaskSet):token = Nonedef setToken(self, token):self.token = tokendef request(self, method, url, **kwargs):headers = kwargs.pop("headers", {})headers.update({"User-Agent": USER_AGENT})if self.token:headers.update({"token": self.token})kwargs.update({'headers': headers, 'catch_response': True})response: ResponseContextManagerwith self.client.request(method, url, **kwargs, ) as response:response.encoding = "utf-8"if response.status_code != 200:response.failure(f"Request {method} {url} failed, status_code: {response.status_code}, text: {response.text}")if response.headers.get("Content-Type", "").startswith("application/json"):js = response.json()if js.get("code") != 2000:response.failure(f"Request {method} {url} failed, response: {js}")return responsedef get(self, url, **kwargs):return self.request("GET", url, **kwargs)def post(self, url, **kwargs):return self.request("POST", url, **kwargs)def put(self, url, **kwargs):return self.request("PUT", url, **kwargs)def delete(self, url, **kwargs):return self.request("DELETE", url, **kwargs)def options(self, url, **kwargs):return self.request("OPTIONS", url, **kwargs)def head(self, url, **kwargs):return self.request("HEAD", url, **kwargs)def patch(self, url, **kwargs):return self.request("PATCH", url, **kwargs)

自定義錯誤

在locust中assert函數只能拋出錯誤只能在Exceptions中查看，這是錯誤，但是用例還是成功的

在這里插入圖片描述

如果標記失敗需要在ResponseContextManager中進行指定，在locust中用例狀態只有兩種，成功，失敗
自定義失敗需要在 ResponseContextManager 中指定一個參數 catch_response= True

例：

若不指定failure，默認用例是成功的，locust中status_code小于400就是成功的用例，不滿足需求常常是自定義的

在這里插入圖片描述

task

項目中一個task便是一個用例，用例必須要有task裝飾器
task有個weight參數，若weight=2，可以簡單的理解為，不指定參數的運行一次，指定參數的運行兩次
例：

# coding: utf-8
from locust import taskfrom common import UserBehaviorBase, operationToken, USER_MANAGEclass InsuranceUser(UserBehaviorBase):"""保險端用戶"""# def on_start(self):#     insurance = Login.login(USER_MANAGE['operation'])#     token = operation.get('result', {}).get('token')#     self.setToken(token)def on_start(self) -> None:self.setToken(operationToken)@taskdef test01(self):"""承保管理 - 詢價申請"""url = '/gateway/spli/insurance/apply/page'payload = {"param": {},"page": {"current": 1,"size": 10}}self.post(url, json=payload)

`on_start/on_stop`

類運行前、后的函數，只運行一次，通常用于登錄、清理內存

執行程序

這里只說一下單使用locust自己寫的用例運行。項目中經常需要在多個文件中寫用例，官方文檔中并沒有明確說該怎么運行多個文件的用例。
其實也比較簡單

例：

# coding: utf-8
from locust import HttpUser, between, run_single_userfrom common import HOST
# 從其他文件中引入類
from tasks import *class WebSite(HttpUser):# 將用例類添加到任務重tasks = [RegulatoryUser, CustomerUser, ProjectUser, InsuranceUser]# tasks = [InsuranceUser]host = HOSTwait_time = between(0.5, 2)if __name__ == '__main__':# 調試模式下，只運行一個用戶run_single_user(WebSite)

參數解析

class WebSite(HttpUser):host = '' # 需要測試的主機地址 eg: http://127.0.0.1:1234tasks: list[TaskSet | Callable] = [] # 任務列表min_wait= 1 # 最小等待時間，已棄用，使用wait_timemax_wait = 2 # 最大等待時間，已棄用 使用 wait_timewait_time=between(3, 25) # 等待時間，這個例子是在2-25秒中隨機選擇一個wait_function=None	# 自定義等待函數，已棄用weight: float = 1	# 與`task`中相同fixed_count: int = 0 # 如果值為> 0，則權重屬性將被忽略，并且將生成'fixed_count'-實例。首先生成這些用戶。如果總目標計數（由——users參數指定）不足以生成具有已定義屬性的每個User類的所有實例，則每個User的最終計數是未定義的。abstract: bool = True	# 如果abstract為True，則該類將被子類化，并且在測試期間不會生成該類的用戶