【Python】萬字長文，Locust 性能測驗指北（上）

Locust

Locust 是比較常見的性能測驗工具，底層基于 gevent，官方介紹 它是一款易于使用、可撰寫腳本且可擴展的性能測驗工具，可以讓我們使用常規 Python 代碼定義用戶的行為，而不必陷入 UI 或限制性領域特定語言中.

Locust具有無限的可擴展性（只要提供客戶端python 代碼，適用于所有協議的性能測驗）.

本文為開發性能自動化對比平臺時學習相關內容的記錄整理，

我們為什么選擇locust

Locust的核心部件

Master節點

負責協調和管理整個測驗程序，包括啟動和停止測驗、分發任務、收集和匯總測驗結果等，

Worker節點

實際執行測驗任務的節點，根據Master節點分配的任務進行模擬用戶行為，

Web UI

提供可視化的測驗界面，方便用戶查看測驗結果、監控測驗進度等，

測驗腳本（Load Test Script）

測驗腳本，定義模擬用戶行為的邏輯和引數，由Worker節點執行，

Locust內部運行呼叫鏈路

時序圖如下：

注：fire() 方法是 Locust 中的 EventHook 類中的一個方法，用于觸發事件，在 Locust 的測驗生命周期中，有多個事件可以被觸發，例如測驗開始、測驗結束、用戶啟動、用戶完成任務等，當這些事件發生時，EventHook 類會呼叫 fire() 方法，將事件傳遞給所有注冊了該事件的回呼函式，

locust 實踐

locust安裝

$ locust -V
locust 2.15.1 from /Users/bingohe/Hebinz/venvnew/lib/python3.9/site-packages/locust (python 3.9.17)

入門示例

使用 locust 撰寫用例時，約定大于配置：

test_xxx (一般測驗框架約定)
dockerfile (docker約定)
locustfile.py (locust約定)

# locustfile.py
from locust import HttpUser, task

class HelloWorldUser(HttpUser): # 父類是個User,表示要生成進行負載測驗的系統的 HTTP“用戶”，每個user相當于一個協程鏈接 ，進行相關系統互動操作
    @task  # 裝飾器來標記為一個測驗任務, 表示用戶要進行的操作:訪問首頁 → 登錄 → 增、刪改查
    def hello_world(self):
        wait_time = between(1, 5)
        # self.client發送 HTTP 請求，模擬用戶的操作
        self.client.get("/helloworld")

啟動測驗

GUI 模式啟動 locust

在有locustfile.py檔案的目錄直接執行locust命令，然后訪問：http://0.0.0.0:8089/ 即可看到下面的界面：

$ locust   
[2023-07-06 16:15:16,868] MacBook-Pro.local/INFO/locust.main: Starting web interface at http://0.0.0.0:8089 (accepting connections from all network interfaces)
[2023-07-06 16:15:16,876] MacBook-Pro.local/INFO/locust.main: Starting Locust 2.15.1

指標詳解：

• Number of users 模擬用戶數，默認 1
• Spawn rate ： 生產數 （每秒）、 =>jmeter : Ramp-Up Period (in seconds)， 默認 1
• Host (e.g. http://www.example.com) => 測驗目標 svr 的 絕對地址

填寫 host點擊 start 之后就會對被測服務如http://{host}/helloworld 發起請求，請求統計資料如下：

WebUI Tab說明：

需要說明的是webui 模式有很多限制，主要用于除錯，下面將要介紹的命令列模式更為常用，

命令列模式啟動 locust

locust -f locustfile.py --headless -u 500 -r 10  --host http://www.example.com  -t 1000s

? 框架是通過命令locust運行的，常用引數有：

檢查點（斷言）

Locust默認情況下會根據HTTP狀態碼來判斷請求是否成功，對于HTTP狀態碼范圍在200-399之間的回應，Locust會將其視為成功，對于HTTP狀態碼在400-599之間的回應，Locust會將其視為失敗，

如果需要根據回應內容或其他條件來判斷請求是否成功，需要手動設定檢查點：

• 使用self.client提供的catch_response=True`引數， 添加locust提供的ResponseContextManager類的背景關系方法手動設定檢查點，
• ResponseContextManager里面的有兩個方法來宣告成功和失敗，分別是success和failure，其中failure方法需要我們傳入一個引數，內容就是失敗的原因，

from locust import HttpUser, task, between

class MyUser(HttpUser):
    # 思考時間：模擬真實用戶在瀏覽應用程式時的行為
    wait_time = between(1, 5)

    @task
    def my_task(self):
        # 基于Locust提供的ResponseContextManager背景關系管理器，使用catch_response=True 引數來捕獲回應，手動標記成功或失敗,
        with self.client.get("/some_page", catch_response=True) as response:
            # 檢查狀態碼是否為200且回應中包含 "some_text"
            if response.status_code == 200 and "some_text" in response.text:
                # 如果滿足條件，標記回應為成功
                response.success()
            else:
                # 如果條件不滿足，根據具體情況生成錯誤資訊
                error_message = "Unexpected status code: " + str(response.status_code) if response.status_code != 200 else "Expected text not found in the response"
                # 標記回應為失敗，并報告錯誤資訊
                response.failure(error_message)

權重比例

如果需要請求有不同的比例，在Locust中，可以通過在@task裝飾器中設定weight引數為任務分配權重來實作，權重越高，任務被執行的頻率就越高，

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(1, 5)

    # 設定權重為3，這個任務將被執行的頻率更高
    @task(3)
    def high_frequency_task(self):
        self.client.get("/high_frequency_page")

    # 設定權重為1，這個任務將被執行的頻率較低
    @task(1)
    def low_frequency_task(self):
        self.client.get("/low_frequency_page")

在這個示例中，我們為high_frequency_task任務設定了權重為3，而為low_frequency_task任務設定了權重為1，這意味著在模擬用戶執行任務時，high_frequency_task任務被執行的頻率將是low_frequency_task任務的3倍，通過設定權重，我們可以根據實際需求調整不同任務在性能測驗中的執行頻率，

import random

class TaskSet:
    def __init__(self):
        self.tasks = []

    def add_task(self, task, weight=1):
        for _ in range(weight):
            self.tasks.append(task)

    def get_random_task(self):
        return random.choice(self.tasks)

task_set = TaskSet()
task_set.add_task("high_frequency_task", weight=3)
task_set.add_task("low_frequency_task", weight=1)

# 當我們呼叫 get_random_task() 方法時，權重較高的任務有更高的概率被選中
random_task = task_set.get_random_task()

引數化

在現實世界中，用戶的行為通常是多樣化的，他們可能使用不同的設備、作業系統、網路條件等，為了更好地模擬這些場景，我們需要在測驗中使用不同的引數，

在性能測驗中，引數化是一種非常重要的技術手段，它允許我們使用不同的資料集運行相同的測驗場景，從而更好地模擬真實世界的用戶行為，常用的引數化方法有兩種，

使用 Locust 的內置引數化功能

from locust import HttpUser, task, between
from locust.randoms import random_string, random_number

class MyUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def random_data(self):
        random_str = random_string(10)
        random_num = random_number(0, 100)
        self.client.post("/random", json={"text": random_str, "number": random_num})

從外部檔案讀取引數

以已經配置成白名單的鑒權 session 為例：

import csv
from locust import HttpUser, task


class CSVReader:
    def __init__(self, file, **kwargs):
        try:
            file = open(file)
        except TypeError:
            pass
        self.file = file
        self.reader = csv.reader(file, **kwargs)  # iterator

    def __next__(self):
        try:
            return next(self.reader)
        except StopIteration:
            # 如果沒有下一行，則從頭開始讀
            self.file.seek(0, 0)
            return next(self.reader)


session = CSVReader("session.csv")

class MyUser(HttpUser):
    @task
    def index(self):
        customer = next(ssn_reader)
        self.client.get(f"/pay?session={customer[0]}")

Tag

在Locust中，標簽（Tag）是用于對任務進行分類和篩選的一種方法，通過給任務添加標簽，可以在運行Locust時只執行具有特定標簽的任務，這在執行特定場景的性能測驗或組織大量任務時非常有用，

使用場景

有時候我們會在同一個檔案中寫多個測驗場景，但是運行的時候只想運行其中一部分，即當一個測驗檔案中的task不止一個時，我們可以通過@tag給task打標簽進行分類，在執行測驗時，通過--tags name執行指定帶標簽的task，

以下是一個使用標簽的示例：

from locust import HttpUser, task, between, tag

class MyUser(HttpUser):
    wait_time = between(1, 5)

    # 給任務添加一個名為 "login" 的標簽
    @tag("login")
    @task
    def login_task(self):
        self.client.post("/login", json={"username": "user", "password": "pass"})

    # 給任務添加一個名為 "profile" 的標簽
    @tag("profile")
    @task
    def profile_task(self):
        self.client.get("/profile")

    # 給任務添加兩個標簽："shopping" 和 "checkout"
    @tag("shopping", "checkout")
    @task
    def checkout_task(self):
        self.client.post("/checkout")

在這個示例中，我們為三個任務分別添加了不同的標簽，login_task任務具有"login"標簽，profile_task任務具有"profile"標簽，而checkout_task任務具有"shopping"和"checkout"兩個標簽，

運行Locust時，可以通過使用--tags和--exclude-tags選項來指定要執行或排除的標簽，例如，要僅執行具有"login"標簽的任務，可以運行：

locust --tags login

要排除具有"shopping"標簽的任務，可以運行：

locust --exclude-tags shopping

這樣，我們就可以根據需要執行特定場景的性能測驗，而不需要修改代碼，

from locust import HttpUser, task, between, tag

@tag("login")
class LoginTasks(HttpUser):
    wait_time = between(1, 5)

    @task
    def login_task(self):
        self.client.post("/login", json={"username": "user", "password": "pass"})

@tag("profile")
class ProfileTasks(HttpUser):
    wait_time = between(1, 5)

    @task
    def profile_task(self):
        self.client.get("/profile")

@tag("shopping", "checkout")
class CheckoutTasks(HttpUser):
    wait_time = between(1, 5)

    @task
    def checkout_task(self):
        self.client.post("/checkout")

集合點

什么是集合點？

集合點用以同步虛擬用戶，以便恰好在同一時刻執行任務，在[測驗計劃]中，可能會要求系統能夠承受1000 人同時提交資料，可以通過在提交資料操作前面加入集合點，這樣當虛擬用戶運行到提交資料的集合點時，就檢查同時有多少用戶運行到集合點，如果不到1000 人，已經到集合點的用戶在此等待，當在集合點等待的用戶達到1000 人時，1000 人同時去提交資料，從而達到測驗計劃中的需求，

注意：Locust框架本身沒有直接封裝集合點的概念 ，需要間接通過gevent并發機制，使用gevent的鎖來實作，

在 Locust 中 實作集合點前，我們先了解兩個概念：

• gevent 中的 Semaphore 信號量
• locust 中的事件鉤子 all_locusts_spawned

Semaphore

信號量（Semaphore）是一種用于控制對共享資源訪問的同步原語，它是計算機科學和并發編程中的一個重要概念，最早由著名計算機科學家Edsger Dijkstra于1960s提出，信號量用于解決多執行緒或多行程環境中的臨界區問題，以防止對共享資源的競爭訪問，

from gevent.lock import Semaphore

# 創建一個具有默認初始值（1）的信號量
sem = Semaphore()

# 創建一個具有自定義初始值（5）的信號量
sem_with_initial_value = https://www.cnblogs.com/Detector/archive/2023/07/08/Semaphore(value=5)

sem.acquire()
# 在此處訪問共享資源

# 完成對共享資源的訪問
sem.release()

with sem:
    # 在此處訪問共享資源
# 信號量會在這里自動釋放

all_locusts_spawned 事件

在 Locust 中，事件是一個非常重要的概念，事件允許我們在 Locust 的生命周期中的特定時刻執行自定義的操作，通過監聽和處理這些事件，我們可以擴展 Locust 的功能，以滿足測驗需求，

spawning_complete 是 Locust 中的一個事件，表示所有的 Locust 用戶（user）已經生成完成，當 Locust 開始運行測驗并生成用戶時，它會逐漸創建用戶實體，一旦所有的用戶都被創建，spawning_complete 事件就會被觸發，你可以在這個事件中執行一些特定的操作，例如輸出日志訊息、收集統計資訊或執行其他自定義操作，

要監聽 spawning_complete 事件，你可以使用 locust.events.spawning_complete 事件鉤子，例如：

from locust import events

@events.spawning_complete.add_listener
def on_spawning_complete():
    print("All users have been spawned!")

在這個示例中，當所有的用戶生成完成時，我們會輸出一條訊息 "All users have been spawned!"，你可以根據需要替換為其他操作，

了解完上面兩個概念，接下來我們只需要兩步走：

• 在腳本啟動時，使用all_locust_spawned.acquire() 阻塞行程
• 撰寫一個函式，在 用戶全部創建完成時觸發 all_locust_spawned.release()

示例代碼：

from locust import HttpUser, task, between
from gevent.lock import Semaphore
from locust import events

all_locust_spawned = Semaphore()
all_locust_spawned.acquire()  # 阻塞


class MyUser(HttpUser):
    wait_time = between(1, 1)

    def on_start(self):
        global all_locust_spawned
        all_locust_spawned.wait(3)  # 同步鎖等待時間

    @task
    def task_rendezvous(self):
        self.client.get("/rendezvous")


# 添加集合點事件處理器
@events.spawning_complete.add_listener  # 所有的Locust實體產生完成時觸發
def on_spawning_complete(**_kwargs):
    global all_locust_spawned
    all_locust_spawned.release()

分布式

當我們需要大量的并發用戶，而單個計算機可能無法生成足夠的負載來模擬這種情況時，分布式壓力測驗可以解決這個問題，我們可以通過將壓力測驗分布到多個計算機上來生成更大的負載，并更準確地評估系統的性能，

Locust 的限制

Locust 使用 Python 的 asyncio 庫來實作異步 I/O，這意味著它可以充分利用多核 CPU 的性能，然而，由于 Python 的全域解釋器鎖（GIL）限制，單個 Python 行程無法充分利用多核 CPU，

為了解決這個問題，Locust 支持在單個計算機上運行多個從節點（worker node），這樣可以充分利用多核 CPU 的性能，

當在單臺計算機上運行多個從節點時，每個從節點將運行在一個單獨的行程中，從而避免了 GIL 的限制，這樣，我們可以充分利用多核 CPU 的性能，生成更大的負載，

單機主從模式

注意： slave 的節點數要小于等于本機的處理器數

在單機主從模式下，主節點和從節點都運行在同一臺計算機上，這種模式適用于在本地開發環境中進行壓力測驗，或者在具有多核 CPU 的單臺服務器上進行壓力測驗，以下是在單機主從模式下實作分布式壓力測驗的步驟：

1. 安裝 Locust：在計算機上安裝 Locust，使用 pip install locust 命令進行安裝，

2. 撰寫 Locust 測驗腳本：撰寫一個 Locust 測驗腳本，這個腳本將在主節點和從節點上運行，將此腳本保存為 locustfile.py，

3. 啟動主節點：在計算機上運行 locust --master 命令啟動主節點，監聽默認埠（8089），

4. 啟動從節點：在計算機上運行 locust --worker --master-host 127.0.0.1 命令啟動一個從節點，根據需要，可以啟動多個從節點，

5. 運行分布式壓力測驗：訪問 Locust 的 Web 界面（http://127.0.0.1:8089），開始測驗，

單機模式下，如何讓每個從節點都運行在不同的 CPU 上

taskset -c CORE_NUMBER locust --worker --master-host 127.0.0.1

taskset -c 0 locust --worker --master-host 127.0.0.1

taskset -c 1 locust --worker --master-host 127.0.0.1

多機主從模式

操作與單機模式基本一樣，訪問 Locust 的 Web 界面時訪問的時主節點的地址（http://MASTER_IP_ADDRESS:8089），

因為主節點和從節點之間通過網路通信，因此，在選擇主節點和從節點的計算機時，需要確保它們之間的網路連接暢通，此外，為了獲得準確的測驗結果，務必確保主節點和從節點之間的網路延遲較低，

分布式模式下的命令引數

--expect-workers 引數用于指定主節點期望連接的從節點數量，如果實際連接的從節點數量沒有達到這個值，主節點會繼續等待，直到足夠的從節點連接上來，

在實際運行分布式壓力測驗時，主節點會在 Web 界面上顯示連接的從節點數量，如果實際連接的從節點數量沒有達到 --expect-workers 指定的值，你可以在 Web 界面上看到一個警告訊息，提示你主節點正在等待更多從節點的連接，

docker 運行locust

使用 容器的方式運行 locust 的優勢和缺點都非常明顯：

但是以下這些場景使用 Docker 來運行 Locust 是一個更好的選擇：

1. 分布式壓力測驗：在分布式壓力測驗中，需要在多臺計算機上運行 Locust 主節點和從節點，使用 Docker 可以確保所有節點的運行環境一致，簡化部署程序，

2. 云環境部署：如果你需要在云環境（如 AWS、Azure、GCP 等）中進行壓力測驗，使用 Docker 可以簡化部署程序，并充分利用云平臺提供的容器服務（如 Amazon ECS、Google Kubernetes Engine 等），

3. CI/CD 集成：如果你需要將壓力測驗集成到持續集成/持續部署（CI/CD）流程中，使用 Docker 可以簡化集成程序，許多 CI/CD 工具（如 Jenkins、GitLab CI、Travis CI 等）都支持 Docker 集成，

4. 避免環境沖突：如果你的開發或測驗環境中已經安裝了其他 Python 應用程式，可能會出現依賴項沖突，使用 Docker 可以將 Locust 運行環境與宿主機系統隔離，避免潛在的環境沖突，

5. 團隊協作：在團隊協作程序中，使用 Docker 可以確保每個團隊成員都使用相同的 Locust 運行環境，從而避免因環境差異導致的問題，

具體使用步驟

1. 首先，確保你已經安裝了 Docker，如果尚未安裝，請參考 Docker 官方檔案 以獲取適用于你的作業系統的安裝說明，

2. 撰寫一個 Locust 測驗腳本，例如，創建一個名為 locustfile.py 的檔案，內容如下：

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(1, 5)

    @task
    def my_task(self):
        self.client.get("/")

3. 使用以下命令從 Docker Hub 拉取官方的 Locust 鏡像：

docker pull locustio/locust

4. 使用以下命令在 Docker 中運行 Locust，

docker run --rm -p 8089:8089 -v $PWD:/mnt/locust locustio/locust -f /mnt/locust/locustfile.py --host TARGET_HOST

在這個命令中，我們將當前目錄（包含 locustfile.py 檔案）掛載到 Docker 容器的 /mnt/locust 目錄，然后，我們使用 -f 引數指定要運行的 Locust 測驗腳本，并使用 --host 引數指定目標主機地址，

5. 訪問 Locust 的 Web 界面，在瀏覽器中打開 http://localhost:8089，你將看到 Locust 的 Web 界面，在這里，你可以開始壓力測驗并查看結果，

通過以上步驟，你可以在 Docker 中運行 Locust，無需在本地環境中安裝 Locust，

總之，在需要確保環境一致性、簡化部署程序、集成到 CI/CD 流程、避免環境沖突或團隊協作的場景下，使用 Docker 來運行 Locust 是一個很好的選擇，通過使用 Docker，你可以輕松地在不同的計算機或云環境中運行壓力測驗，從而實作更大規模的分布式壓力測驗，

高性能 FastHttpUser

Locust 的默認 HTTP 客戶端使用http.client，如果計劃以非常高的吞吐量運行測驗并且運行 Locust 的硬體有限，那么它有時效率不夠，

FastHttpUser 是 Locust 提供的一個特殊的用戶類，用于執行 HTTP 請求，與默認的 HttpUser 不同，FastHttpUser 使用 C 語言庫 gatling 撰寫的 httpclient 進行 HTTP 請求, 有時將給定硬體上每秒的最大請求數增加了 5 到 6 倍，在相同的并發條件下使用FastHttpUser能有效減少對負載機的資源消耗從而達到更大的http請求，

優勢

1. 性能：FastHttpUser 的主要優勢是性能，由于它使用 C 語言庫進行 HTTP 請求，它的性能通常比默認的 HttpUser 更高，這意味著在相同的硬體資源下，你可以使用 FastHttpUser 生成更大的負載，

2. 資源占用：與默認的 HttpUser 相比，FastHttpUser 通常具有較低的資源占用（如 CPU、記憶體等），這意味著在進行壓力測驗時，你可以在同一臺計算機上運行更多的并發用戶，

3. 更高的并發能力：由于 FastHttpUser 的性能和資源占用優勢，它可以更好地支持大量并發用戶的壓力測驗，這對于需要模擬大量并發用戶的場景（如高流量 Web 應用程式、API 等）非常有用，

然而需要注意的是FastHttpUser 也有一些局限性，例如，它可能不支持某些特定的 HTTP 功能（如自定義 SSL 證書、代理設定等），在選擇使用 FastHttpUser 時，需要權衡性能優勢和功能支持，如果測驗場景不需要大量并發用戶，或者需要特定的 HTTP 功能，使用默認的 HttpUser 可能更合適，

以下是一個使用 FastHttpUser 的 Locust 測驗腳本示例：

from locust import FastHttpUser, task, between

class MyFastHttpUser(FastHttpUser):
    wait_time = between(1, 5)

    @task
    def my_task(self):
        self.client.get("/")

測驗gRPC等其他協議

locust 并非 http 介面測驗工具 ， 只是內置了 “HttpUser” 示例 ，理論上來說，只要提供客戶端，它可以測驗任何協議，

如果有測驗 gRPC、XML-RPC、requests-based libraries/SDKs等需求，可以參考：
https://docs.locust.io/en/stable/testing-other-systems.html

其他

主流性能測驗工具對比

下面是 Locust、JMeter、Wrk 和 LoadRunner 四款性能測驗工具的優缺點和支持的功能的對比表格：

需要注意的是，這些工具的優缺點和支持的功能只是相對而言的，具體使用時需要根據實際需求和場景選擇，

合抱之木，生于毫末；九層之臺，起于累土；千里之行，始于足下，

標籤：其他

上一篇：AtCoder Beginner Contest 309

下一篇：返回列表