Day05 - 深入淺出 Task 以及 TaskSet 的使用 - iT 邦幫忙::一起幫忙解決難題，拯救 IT 人的一天

2025 iThome 鐵人賽

DAY 5

DevOps

Vibe Coding 後的挑戰：Locust x Loki 負載及監控系列第 5 篇

Day05 - 深入淺出 Task 以及 TaskSet 的使用

17th鐵人賽 python loki locust grafana

熊熊工程師

團隊一蘭拉麵基本配料 5 倍辣

2025-08-24 10:44:28

137 瀏覽

分享至

在前幾天的介紹中，我們已經初步認識了 Locust 的基本用法。今天，我們將深入探討 Locust 中的核心概念：Task 與 TaskSet。透過靈活運用這些功能，您可以更精準地模擬真實使用者的行為模式，讓壓力測試更貼近實際場景。

準備測試環境

在執行本章的範例程式碼前，請確保您的 FastAPI 服務（main.py）已經包含了相應的測試端點。如果您發現某些端點不存在，可以參考以下範例自行添加：

# 基本的測試端點範例
@app.get("/items")
async def get_items():
    return {"items": [{"id": 1, "name": "商品1"}]}

@app.post("/cart")
async def add_to_cart(item: Dict):
    return {"status": "success"}

@app.get("/api/{endpoint}")
async def api_endpoint(endpoint: str):
    return {"endpoint": endpoint, "status": "ok"}

您可以根據測試需求，靈活添加更多端點。確保 FastAPI 服務在 http://localhost:8080 上運行後，即可開始練習本章的範例。

@task 裝飾器與權重設定

在 Locust 中，@task 裝飾器是定義使用者任務的核心機制。每個被 @task 裝飾的方法都代表一個使用者可能執行的操作。當一個 User 類別中定義了多個任務時，Locust 會根據任務的權重來決定執行機率。

基本用法與權重參數

from locust import HttpUser, task, between

class WeightedTaskUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"
    
    @task(3)  # 權重為 3
    def view_items(self):
        """瀏覽商品列表"""
        self.client.get("/items")
    
    @task(2)  # 權重為 2
    def view_item_details(self):
        """查看商品詳情"""
        self.client.get("/items/1")
    
    @task(1)  # 權重為 1
    def add_to_cart(self):
        """加入購物車"""
        self.client.post("/cart", json={"item_id": 1, "quantity": 1})

在上述範例中，三個任務的權重比例為 3:2:1，這意味著：

view_items 有 50% 的機率被執行 (3/6)
view_item_details 有 33.3% 的機率被執行 (2/6)
add_to_cart 有 16.7% 的機率被執行 (1/6)

讓我們使用無頭模式執行這個範例，觀察實際的執行分布：

locust -f task_weight_demo.py --headless --users 10 --spawn-rate 1 --run-time 20s --host http://localhost:8080

執行結果顯示，在 20 秒的測試中：

Type     Name                          # reqs      # fails |    Avg     Min     Max    Med |   req/s
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
POST     /cart                              17     0(0.00%) |      2       1       9      2 |    0.87
GET      /items                             43     0(0.00%) |      2       1       6      2 |    2.20
GET      /items/1                           21     0(0.00%) |      2       1       5      2 |    1.08
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
         Aggregated                         81     0(0.00%) |      2       1       9      2 |    4.15

從結果可以看出：

/items 被呼叫 43 次 (53.1%)
/items/1 被呼叫 21 次 (25.9%)
/cart 被呼叫 17 次 (21.0%)

這個分布大致符合我們設定的 3:2:1 權重比例

未設定權重的隨機執行

如果不指定權重參數，所有任務的預設權重都是 1，每個任務被選中的機率相等：

from locust import HttpUser, task, between
import random

class RandomTaskUser(HttpUser):
    wait_time = between(1, 2)
    host = "http://localhost:8080"
    
    @task  # 未指定權重，預設為 1
    def task_a(self):
        print("執行任務 A")
        self.client.get("/api/a")
    
    @task  # 未指定權重，預設為 1
    def task_b(self):
        print("執行任務 B")
        self.client.get("/api/b")
    
    @task  # 未指定權重，預設為 1
    def task_c(self):
        print("執行任務 C")
        self.client.get("/api/c")

在這個例子中，每個任務都有 33.3% 的機率被執行。每次虛擬使用者需要執行任務時，都會隨機選擇一個任務執行。

執行無頭模式測試：

locust -f task_random_demo.py --headless --users 3 --spawn-rate 1 --run-time 10s --host http://localhost:8080

執行結果與終端輸出：

執行任務 C
執行任務 A
執行任務 B
執行任務 A
執行任務 B
執行任務 C
...

Type     Name                          # reqs      # fails |    Avg     Min     Max    Med |   req/s
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
GET      /api/a                              7     0(0.00%) |      2       1       3      2 |    0.72
GET      /api/b                              6     0(0.00%) |      2       1       2      2 |    0.62
GET      /api/c                              7     0(0.00%) |      2       1       8      2 |    0.72
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
         Aggregated                         20     0(0.00%) |      2       1       8      2 |    2.07

可以看到三個任務的執行次數非常接近（7、6、7），證明了在未設定權重時，每個任務的執行機率確實相等。

透過 tasks 參數指定任務

除了使用 @task 裝飾器外，Locust 還提供了另一種定義任務的方式：透過 tasks 屬性。這種方式在某些場景下更為靈活。

使用列表方式定義任務

from locust import HttpUser, between

def browse_products(user):
    user.client.get("/products")

def search_product(user):
    user.client.get("/search?q=laptop")

def view_product(user):
    user.client.get("/products/123")

class ListTaskUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"
    
    # 使用列表定義任務，每個任務權重相等
    tasks = [browse_products, search_product, view_product]

使用字典方式定義任務並設定權重

from locust import HttpUser, between

def homepage(user):
    user.client.get("/")

def browse_catalog(user):
    user.client.get("/catalog")

def user_profile(user):
    user.client.get("/profile")

def checkout(user):
    user.client.post("/checkout")

class DictTaskUser(HttpUser):
    wait_time = between(2, 5)
    host = "http://localhost:8080"
    
    # 使用字典定義任務並設定權重
    tasks = {
        homepage: 4,         # 40% 機率
        browse_catalog: 3,   # 30% 機率
        user_profile: 2,     # 20% 機率
        checkout: 1          # 10% 機率
    }

使用字典方式的優勢在於可以更清楚地看到每個任務的權重設定，特別適合管理大量任務的場景。

TaskSet 的使用與組合

TaskSet 是 Locust 提供的任務組織機制，讓您可以將相關的任務組合在一起，更好地模擬使用者的行為流程，讓我們看一下線面這段程式：

TaskSet 當中同樣也可以使用 on_start 以及 on_stop 來管理任務流程

from locust import HttpUser, TaskSet, task, between

class UserBehavior(TaskSet):
    """定義使用者行為的 TaskSet"""
    
    @task(2)
    def index_page(self):
        self.client.get("/")
    
    @task(3)
    def browse_products(self):
        self.client.get("/products")
    
    @task(1)
    def about_page(self):
        self.client.get("/about")
    
    def on_start(self):
        """TaskSet 開始時執行"""
        print("使用者開始瀏覽網站")
        # 模擬登入
        self.client.post("/login", json={
            "username": "test_user",
            "password": "password123"
        })

class WebsiteUser(HttpUser):
    wait_time = between(1, 5)
    host = "http://localhost:8080"
    tasks = [UserBehavior]  # 將 TaskSet 指派給 User

執行這個 TaskSet 範例：

locust -f taskset_basic_demo.py --headless --users 5 --spawn-rate 1 --run-time 15s --host http://localhost:8080

執行結果顯示了 TaskSet 的行為模式：

使用者開始瀏覽網站
使用者開始瀏覽網站
使用者開始瀏覽網站
使用者開始瀏覽網站
使用者開始瀏覽網站

Type     Name                          # reqs      # fails |    Avg     Min     Max    Med |   req/s
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
GET      /                                   7     0(0.00%) |      2       1       2      2 |    0.48
GET      /about                              6     0(0.00%) |      1       1       2      2 |    0.41
POST     /login                              5     0(0.00%) |      4       2       9      4 |    0.34
GET      /products                          13     0(0.00%) |      1       1       2      2 |    0.89
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
         Aggregated                         31     0(0.00%) |      2       1       9      2 |    2.12

可以看到：

每個使用者在開始時都執行了 on_start() 方法進行登入
根據權重設定，/products (權重3) 的請求次數最多
index_page (權重2) 和 about_page (權重1) 的執行次數符合預期比例

SequentialTaskSet 的使用

當您需要模擬有順序的使用者行為時，SequentialTaskSet 是最佳選擇。它會按照定義的順序依次執行任務，非常適合模擬具有固定流程的場景。

使用情境與實作

from locust import HttpUser, SequentialTaskSet, task, between

class UserRegistrationFlow(SequentialTaskSet):
    """模擬使用者註冊流程"""
    
    @task
    def visit_homepage(self):
        """步驟 1: 訪問首頁"""
        print("訪問首頁")
        self.client.get("/")
    
    @task
    def go_to_signup(self):
        """步驟 2: 前往註冊頁面"""
        print("前往註冊頁面")
        self.client.get("/signup")
    
    @task
    def submit_registration(self):
        """步驟 3: 提交註冊表單"""
        print("提交註冊表單")
        response = self.client.post("/api/register", json={
            "username": f"user_{self.user.username_counter}",
            "email": f"user_{self.user.username_counter}@example.com",
            "password": "secure_password123"
        })
        
        if response.status_code == 200:
            self.user.username_counter += 1
    
    @task
    def verify_email(self):
        """步驟 4: 驗證電子郵件"""
        print("驗證電子郵件")
        self.client.post("/api/verify-email", json={
            "token": "mock_verification_token"
        })
    
    @task
    def complete_profile(self):
        """步驟 5: 完成個人資料"""
        print("完成個人資料設定")
        self.client.post("/api/profile", json={
            "full_name": "Test User",
            "bio": "This is a test user"
        })
    
    @task
    def explore_features(self):
        """步驟 6: 探索功能"""
        print("開始探索網站功能")
        self.client.get("/dashboard")
        # 完成註冊流程後，可以中斷此 TaskSet
        self.interrupt()

class NewUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"
    tasks = [UserRegistrationFlow]
    
    def on_start(self):
        """初始化使用者計數器"""
        self.username_counter = 1

執行 SequentialTaskSet 的範例：

locust -f sequential_taskset_demo.py --headless --users 2 --spawn-rate 1 --run-time 10s --host http://localhost:8080

執行輸出顯示了任務的順序執行：

訪問首頁
訪問首頁
前往註冊頁面
前往註冊頁面
提交註冊表單
提交註冊表單
驗證電子郵件
驗證電子郵件
完成個人資料設定

Type     Name                          # reqs      # fails |    Avg     Min     Max    Med |   req/s
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
GET      /                                   2     0(0.00%) |      8       2      13      3 |    0.22
POST     /api/profile                        1     0(0.00%) |      6       6       6      6 |    0.11
POST     /api/register                       2     0(0.00%) |      2       2       2      2 |    0.22
POST     /api/verify-email                   2     0(0.00%) |      2       2       2      2 |    0.22
GET      /signup                             2     0(0.00%) |      2       2       2      2 |    0.22

可以清楚地看到：

任務按照定義的順序依次執行
每個使用者都完整地走過了註冊流程的每一步
不同於普通 TaskSet，這裡沒有隨機性，任務嚴格按順序執行

複雜的順序流程範例

from locust import HttpUser, SequentialTaskSet, task, between
import random

class PurchaseFlow(SequentialTaskSet):
    """模擬完整的購買流程"""
    
    def on_start(self):
        """初始化購買流程所需的資料"""
        self.selected_products = []
        self.order_id = None
    
    @task
    def search_products(self):
        """搜尋商品"""
        search_terms = ["laptop", "phone", "tablet", "headphones"]
        term = random.choice(search_terms)
        print(f"搜尋: {term}")
        response = self.client.get(f"/api/search?q={term}")
        # 假設回應包含商品列表
        self.search_results = response.json().get("products", [])
    
    @task
    def browse_search_results(self):
        """瀏覽搜尋結果"""
        if hasattr(self, 'search_results') and self.search_results:
            # 隨機選擇 1-3 個商品查看詳情
            num_products = min(len(self.search_results), random.randint(1, 3))
            for _ in range(num_products):
                product = random.choice(self.search_results)
                self.client.get(f"/api/products/{product['id']}")
                # 50% 機率將商品加入待購清單
                if random.random() > 0.5:
                    self.selected_products.append(product)
    
    @task
    def add_to_cart(self):
        """將選中的商品加入購物車"""
        if self.selected_products:
            for product in self.selected_products:
                print(f"加入購物車: {product['name']}")
                self.client.post("/api/cart/add", json={
                    "product_id": product['id'],
                    "quantity": random.randint(1, 2)
                })
    
    @task
    def review_cart(self):
        """檢視購物車"""
        print("檢視購物車內容")
        self.client.get("/api/cart")
    
    @task
    def proceed_to_checkout(self):
        """進入結帳頁面"""
        if self.selected_products:
            print("進入結帳流程")
            self.client.get("/checkout")
        else:
            print("購物車為空，結束流程")
            self.interrupt()
    
    @task
    def enter_shipping_info(self):
        """輸入配送資訊"""
        print("輸入配送資訊")
        self.client.post("/api/checkout/shipping", json={
            "address": "123 Test Street",
            "city": "Test City",
            "postal_code": "12345"
        })
    
    @task
    def select_payment_method(self):
        """選擇付款方式"""
        print("選擇付款方式")
        self.client.post("/api/checkout/payment", json={
            "method": "credit_card",
            "card_number": "4111111111111111"
        })
    
    @task
    def confirm_order(self):
        """確認訂單"""
        print("確認訂單")
        response = self.client.post("/api/checkout/confirm")
        if response.status_code == 200:
            self.order_id = response.json().get("order_id")
            print(f"訂單完成！訂單編號: {self.order_id}")
    
    @task
    def view_order_confirmation(self):
        """查看訂單確認頁面"""
        if self.order_id:
            print(f"查看訂單確認: {self.order_id}")
            self.client.get(f"/orders/{self.order_id}")

class ShoppingUser(HttpUser):
    wait_time = between(2, 4)
    host = "http://localhost:8080"
    tasks = [PurchaseFlow]

SequentialTaskSet 特別適合以下使用情境：

註冊流程：從填寫表單到驗證郵件的完整流程
購物流程：從搜尋商品到完成結帳的步驟
多步驟表單：需要按順序填寫的複雜表單
教學導覽：模擬新使用者的引導流程

NestedTaskSet 巢狀任務集

在複雜的應用場景中，您可能需要在 TaskSet 中嵌套其他 TaskSet，以更精確地模擬使用者的行為模式。

基本巢狀結構

from locust import HttpUser, TaskSet, task, between
import random

class BrowsingBehavior(TaskSet):
    """瀏覽行為子任務集"""
    
    @task(3)
    def view_product_list(self):
        category = random.choice(["electronics", "clothing", "books"])
        print(f"瀏覽 {category} 分類")
        self.client.get(f"/categories/{category}")
    
    @task(2)
    def search_products(self):
        keyword = random.choice(["laptop", "shirt", "python"])
        print(f"搜尋: {keyword}")
        self.client.get(f"/search?q={keyword}")
    
    @task(1)
    def view_deals(self):
        print("查看特價商品")
        self.client.get("/deals")
    
    @task(1)
    def stop_browsing(self):
        """結束瀏覽，返回父任務集"""
        print("結束瀏覽行為")
        self.interrupt()

class AccountManagement(TaskSet):
    """帳戶管理子任務集"""
    
    @task(2)
    def view_profile(self):
        print("查看個人資料")
        self.client.get("/profile")
    
    @task(2)
    def view_orders(self):
        print("查看歷史訂單")
        self.client.get("/orders")
    
    @task(1)
    def update_settings(self):
        print("更新帳戶設定")
        self.client.put("/settings", json={
            "notifications": random.choice([True, False]),
            "newsletter": random.choice([True, False])
        })
    
    @task(1)
    def exit_account(self):
        """離開帳戶管理"""
        print("離開帳戶管理")
        self.interrupt()

class MainUserBehavior(TaskSet):
    """主要使用者行為任務集"""
    
    # 嵌套多個子任務集
    tasks = {
        BrowsingBehavior: 3,    # 60% 機率進入瀏覽模式
        AccountManagement: 1    # 20% 機率進入帳戶管理
    }
    
    @task(1)  # 20% 機率執行單一任務
    def check_notifications(self):
        print("檢查通知")
        self.client.get("/notifications")

class NestedTaskUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"
    tasks = [MainUserBehavior]

巢狀 TaskSet 適合以下場景：

多平台模擬：模擬使用者在不同平台（網頁、手機 APP）間切換
角色區分：不同類型的使用者有不同的行為模式
功能模組化：將複雜的功能拆分成獨立的行為模組
狀態轉換：模擬使用者在不同狀態間的轉換

需要特別注意的是，當設定了多個 TaskSet 時，測試時的虛擬使用者會隨機挑選要執行哪個 TaskSet。如果想要更精確地控制這個行為，可以：

透過設定權重來調整不同 TaskSet 被選中的機率
使用 self.interrupt() 方法來主動切換 TaskSet
重新設計 TaskSet 的結構，使其更符合實際的使用者行為

self.interrupt() 方法的使用

self.interrupt() 是 TaskSet 中一個重要的控制方法，它允許您中斷當前的 TaskSet 執行，返回到父級 TaskSet 或重新選擇任務。下面簡單列出這個方法的典型使用情境：

登入/登出流程：成功登入後離開登入 TaskSet
錯誤處理：遇到錯誤時中斷當前操作
條件判斷：根據特定條件（如餘額、庫存）決定是否繼續
時間控制：限時任務或會話超時處理
優先級管理：發現更重要的任務時中斷當前任務
狀態轉換：完成特定狀態後轉換到其他行為模式

下面是一個搭配 self.interrupt() 撰寫而成的 TaskSet 範例，可以看到當符合某些條件後，可以直接在 funciont 當中調用 self.interrupt() 來中斷這個任務流程：

from locust import HttpUser, TaskSet, task, between
import random

class LoginBehavior(TaskSet):
    """登入相關行為"""
    
    def on_start(self):
        self.login_attempts = 0
    
    @task
    def attempt_login(self):
        self.login_attempts += 1
        print(f"嘗試登入 (第 {self.login_attempts} 次)")
        
        response = self.client.post("/api/login", json={
            "username": "test_user",
            "password": "password123"
        })
        
        if response.status_code == 200:
            print("登入成功！")
            # 登入成功後中斷此 TaskSet
            self.interrupt()
        elif self.login_attempts >= 3:
            print("登入失敗次數過多，暫時放棄")
            self.interrupt()

class SessionBehavior(TaskSet):
    """使用者會話行為"""
    
    def on_start(self):
        self.action_count = 0
        self.max_actions = random.randint(5, 10)
    
    @task(3)
    def browse_content(self):
        self.action_count += 1
        print(f"瀏覽內容 ({self.action_count}/{self.max_actions})")
        self.client.get("/api/content")
        
        # 達到預定操作次數後結束會話
        if self.action_count >= self.max_actions:
            print("完成預定操作，結束會話")
            self.interrupt()
    
    @task(1)
    def check_messages(self):
        print("檢查訊息")
        response = self.client.get("/api/messages")
        
        # 如果有緊急訊息，中斷當前行為處理
        if response.json().get("urgent_count", 0) > 0:
            print("發現緊急訊息，中斷當前任務")
            self.interrupt()

class ConditionalTaskSet(TaskSet):
    """條件式任務集"""
    
    def on_start(self):
        self.balance = 1000
    
    @task(3)
    def normal_transaction(self):
        amount = random.randint(10, 100)
        self.balance -= amount
        print(f"執行交易: -{amount}, 餘額: {self.balance}")
        
        self.client.post("/api/transaction", json={
            "type": "purchase",
            "amount": amount
        })
        
        # 餘額不足時中斷
        if self.balance < 100:
            print("餘額不足，需要充值")
            self.interrupt()
    
    @task(1)
    def check_balance(self):
        print(f"查詢餘額: {self.balance}")
        self.client.get("/api/balance")
        
        # 餘額過低時觸發充值流程
        if self.balance < 50:
            print("餘額過低，立即中斷進行充值")
            self.interrupt()

class UserWithInterrupts(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"
    
    tasks = {
        LoginBehavior: 1,
        SessionBehavior: 3,
        ConditionalTaskSet: 2
    }

執行帶有 interrupt 的範例：

locust -f interrupt_demo.py --headless --users 3 --spawn-rate 1 --run-time 15s --host http://localhost:8080

執行過程的輸出片段：

瀏覽內容 (1/10)
瀏覽內容 (1/6)
執行交易: -25, 餘額: 975
瀏覽內容 (2/6)
瀏覽內容 (2/10)
查詢餘額: 975
檢查訊息
瀏覽內容 (3/6)
...
執行交易: -63, 餘額: 912
執行交易: -18, 餘額: 894
瀏覽內容 (6/6)
完成預定操作，結束會話  # interrupt() 被觸發
嘗試登入 (第 1 次)

Type     Name                          # reqs      # fails |    Avg     Min     Max    Med |   req/s
--------|------------------------------|-------|-------------|-------|-------|-------|-------|--------
GET      /api/balance                        4     0(0.00%) |      3       1       6      2 |    0.29
GET      /api/content                       11     0(0.00%) |      5       1      11      4 |    0.80
GET      /api/messages                       5     0(0.00%) |      2       1       5      2 |    0.36
POST     /api/transaction                    3     0(0.00%) |      2       2       3      2 |    0.22

從輸出可以看到 interrupt() 方法的實際應用：