如何在Locust中实现Transaction Controller及测量每秒完成任务数

有两种常用方法可以实现这个需求，核心是把多请求的流程作为一个整体来统计完成率和每秒处理数：

方法1：手动利用事件钩子封装事务 #

将多请求流程包裹起来，通过Locust的events模块触发成功/失败事件，把整个流程标记为一个"事务"，这样就能在统计面板中看到每秒完成的流程数。

from locust import HttpUser, task, events
import time

class WorkflowUser(HttpUser):
    @task
    def complete_user_flow(self):
        start_time = time.time()
        try:
            # 执行你的多请求业务流程
            self.client.get("/login-page")
            self.client.post("/login", json={"username": "test", "password": "123"})
            self.client.get("/user-profile")
            self.client.post("/update-profile", json={"bio": "new bio"})
            
            # 记录事务成功
            elapsed_ms = int((time.time() - start_time) * 1000)
            events.request_success.fire(
                request_type="TRANSACTION",
                name="user_profile_update_flow",
                response_time=elapsed_ms,
                response_length=0
            )
        except Exception as e:
            # 记录事务失败
            elapsed_ms = int((time.time() - start_time) * 1000)
            events.request_failure.fire(
                request_type="TRANSACTION",
                name="user_profile_update_flow",
                response_time=elapsed_ms,
                exception=e
            )

方法2：用上下文管理器复用事务逻辑 #

把事务统计逻辑抽离成上下文管理器，让代码更简洁易复用：

from locust import HttpUser, task, events
import time
from contextlib import contextmanager

@contextmanager
def transaction(name):
    start_time = time.time()
    try:
        yield
        # 事务成功时触发事件
        elapsed_ms = int((time.time() - start_time) * 1000)
        events.request_success.fire(
            request_type="TRANSACTION",
            name=name,
            response_time=elapsed_ms,
            response_length=0
        )
    except Exception as e:
        # 事务失败时触发事件
        elapsed_ms = int((time.time() - start_time) * 1000)
        events.request_failure.fire(
            request_type="TRANSACTION",
            name=name,
            response_time=elapsed_ms,
            exception=e
        )

class WorkflowUser(HttpUser):
    @task
    def checkout_flow(self):
        with transaction("checkout_process"):
            self.client.get("/cart")
            self.client.post("/apply-coupon", json={"code": "SAVE10"})
            self.client.post("/place-order", json={"shipping": "standard"})
            self.client.get("/order-confirmation")

查看统计结果 #

• 在Locust的Web UI的Statistics标签页中，你会看到类型为TRANSACTION的统计项，其中Req/s就是每秒完成的多请求流程数，和JMeter事务控制器的TPS指标逻辑完全一致。
• 命令行模式下生成的CSV报告也会包含该事务的完整统计数据，可用于后续分析。

内容的提问来源于stack exchange，提问作者Moaad Shbita