官方文檔：https://docs.locust.io/en/stable/

基礎demo

from locust import HttpUser, task


class HelloWorldUser(HttpUser):
    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

最佳實踐

• 斷言：
  在 Locust 中，默認情況下，如果 HTTP 響應狀態碼是 2xx，請求會被標記為成功；否則會被標記為失敗

• SequentialTaskSet和TaskSet的區別：
  TaskSet 是 Locust 的一個基本類，用于定義用戶行為。
  SequentialTaskSet 是 TaskSet 的子類，它會按照任務定義的順序依次執行任務，每個任務執行完后才會執行下一個任務。

• task：
  方法上添加 @task 裝飾器來定義用戶任務。

• 定義負載形狀：

  1. LoadTestShape自動化控制（使用該方式將無法通過WebUI手動控制）
     用于定義負載形狀，即用戶數量和請求頻率的變化規律?？梢酝ㄟ^繼承 LoadTestShape 類并重寫 tick 方法來自定義負載形狀。
     tick 方法會在每個時間間隔內被調用，返回一個元組 (users, spawn_rate)，表示當前的用戶數量和每秒啟動用戶數。
  2. 使用 WebUI 手動調整
     在壓測過程中，隨時點擊右上角的 "+" 或 "-" 按鈕 增加 / 減少用戶數。

• HttpUser：
  用于定義 HTTP 用戶類.每個 HttpUser 實例代表一個用戶，它會執行定義的任務。

  1. host: 指定請求的基礎URL
  2. wait_time: 指定用戶任務之間的等待時間
  3. tasks: 指定用戶的任務列表

• name:
  對接口請求進行分組，接口參數可能是動態的，在用戶統計信息中將這些 URL 組合在一起才有意義。

• 鉤子函數：

  1. on_test_start: 測試開始時執行的函數
  2. on_test_stop: 測試結束時執行的函數

from locust import HttpUser, task, between, LoadTestShape, SequentialTaskSet, constant_throughput, events


@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """
    測試開始時執行的函數,
    """
    print("測試開始")


class UserBehavior(SequentialTaskSet):
    """用戶行為"""
    token = None

    def on_start(self):
        """每個用戶啟動時執行的函數"""
        print("用戶啟動")

    @task
    def login(self):
        """登錄"""
        with self.client.post("/login", name="登錄", catch_response=True) as response:
            if response.status_code == 200:
                self.token = response.json().get("token")
                response.success()
            else:
                response.failure("status_code非200：" + response.text)

    @task(3)
    def create_product(self):
        """創建商品"""
        if not self.token:
            return  # 如果沒有token，不執行創建商品任務

        body = {
            "productName": "測試商品",
        }
        headers = {
            "Authorization": f"Bearer {self.token}"
        }
        with self.client.post(url="/create_product", name="創建商品", json=body,headers=headers,
                              catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure("status_code非200：" + response.text)

    @task
    def logout(self):
        """退出登錄"""
        if not self.token:
            return  # 如果沒有token，不執行退出登錄任務
        with self.client.post("/logout", name="退出登錄", catch_response=True) as response:
            if response.status_code == 200:
                self.token = None  # 登出后清空token
                response.success()
            else:
                response.failure("status_code非200：" + response.text)


class TimeLimitLoadShape(LoadTestShape):
    """10分鐘后自動退出的負載形狀"""

    def tick(self):
        # 獲取當前壓測運行時間（秒）
        run_time = self.get_run_time()
        # 10分鐘 = 600秒，超過則退出
        if run_time > 600:
            return None  # 返回None表示終止測試
        # 壓測期間保持100用戶（可根據需求調整用戶數和增長策略）
        return (100, 10)  # (目標用戶數, 每秒啟動用戶數)


class LiveAuctionUser(HttpUser):
    tasks = [UserBehavior]
    wait_time = constant_throughput(2)  # 每個用戶每秒最多執行2次循環

負載壓測實踐

class StepLoadShape(LoadTestShape):
    """
    階梯壓測形狀：
    - 每 step_duration 秒增加 step_users 個用戶
    - 直到達到 max_users 為止
    """
    step_time = 10  # 每個階梯持續時間（秒）
    step_users = 10  # 每個階梯增加的用戶數
    spawn_rate = 10  # 每秒啟動的用戶數
    max_users = 100  # 最大用戶數

    def tick(self):
        # 計算當前所處的階梯數
        run_time = self.get_run_time()
        current_step = int(run_time / self.step_time) + 1

        # 計算當前應有的用戶數
        users = current_step * self.step_users

        # 檢查是否達到最大用戶數
        if users > self.max_users:
            return None  # 停止壓測

        return (users, self.spawn_rate)

class StagedLoadShape(LoadTestShape):
    """
    分階段階梯壓測：
    1. 預熱階段
    2. 穩定階段
    3. 加壓階段
    4. 收尾階段
    """
    stages = [
        {"duration": 10, "users": 50, "spawn_rate": 10},  # 預熱階段
        {"duration": 20, "users": 50, "spawn_rate": 10},  # 穩定階段
        {"duration": 30, "users": 100, "spawn_rate": 10},  # 加壓階段1
        {"duration": 40, "users": 150, "spawn_rate": 10},  # 加壓階段2
        {"duration": 45, "users": 100, "spawn_rate": 20},  # 減壓階段1
        {"duration": 50, "users": 50, "spawn_rate": 20},  # 減壓階段2
        {"duration": 55, "users": 0, "spawn_rate": 20},  # 減壓階段3
    ]

    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

高階使用

1. 分布式測試
2. 數據參數化
3. TaskSet嵌套