如何按任务分组Locust统计数据？内部库发起请求无法用name参数

针对你遇到的两个任务共用response/{id}/端点、无法通过name参数分组统计的问题，这里提供几个无需修改内部库的解决思路：

方法1：利用Locust事件钩子标记任务 #

通过监听请求成功/失败事件，结合任务执行时的标记，给请求统计名称加上任务前缀，实现分组统计。

from locust import HttpUser, task, events

class MyUser(HttpUser):
    # 记录当前执行的任务名
    current_task = None

    @task
    def fetch_task(self):
        self.current_task = "fetch"
        # 调用内部库发起请求
        internal_lib.fetch(self.client)
        self.current_task = None

    @task
    def update_task(self):
        self.current_task = "update"
        internal_lib.update(self.client)
        self.current_task = None

# 处理请求成功事件，修改统计名称
@events.request_success.add_listener
def rewrite_success_stats(request_type, name, response_time, response_length, **kwargs):
    user = kwargs.get("user")
    if user and user.current_task:
        # 生成带任务前缀的新统计名称
        new_name = f"{user.current_task}:{name}"
        # 用新名称触发统计，同时阻止原统计记录
        events.request_success.fire(
            request_type=request_type,
            name=new_name,
            response_time=response_time,
            response_length=response_length,
            **kwargs
        )
        return False

# 同理处理请求失败事件
@events.request_failure.add_listener
def rewrite_failure_stats(request_type, name, response_time, exception, **kwargs):
    user = kwargs.get("user")
    if user and user.current_task:
        new_name = f"{user.current_task}:{name}"
        events.request_failure.fire(
            request_type=request_type,
            name=new_name,
            response_time=response_time,
            exception=exception,
            **kwargs
        )
        return False

效果：统计面板会显示fetch:response/{id}/和update:response/{id}/两个独立的条目，分别对应不同任务下的请求耗时。

方法2：通过自定义HTTP适配器打标记 #

如果内部库基于requests实现，可以给Locust的client挂载自定义适配器，在请求发送时标记任务名，再通过事件钩子修改统计名称。

from locust import HttpUser, task, events
from requests.adapters import HTTPAdapter

class TaskMarkAdapter(HTTPAdapter):
    def __init__(self, task_name, *args, **kwargs):
        self.task_name = task_name
        super().__init__(*args, **kwargs)

    def send(self, request, **kwargs):
        # 给请求对象添加任务名标记
        request.task_name = self.task_name
        return super().send(request, **kwargs)

class MyUser(HttpUser):
    @task
    def fetch_task(self):
        # 挂载带任务标记的适配器
        adapter = TaskMarkAdapter("fetch")
        self.client.mount("http://", adapter)
        self.client.mount("https://", adapter)
        
        internal_lib.fetch(self.client)
        
        # 任务结束后移除适配器，避免影响其他任务
        self.client.unmount("http://")
        self.client.unmount("https://")

    @task
    def update_task(self):
        adapter = TaskMarkAdapter("update")
        self.client.mount("http://", adapter)
        self.client.mount("https://", adapter)
        
        internal_lib.update(self.client)
        
        self.client.unmount("http://")
        self.client.unmount("https://")

# 监听请求事件，根据标记修改统计名称
@events.request_success.add_listener
def adapt_success_stats(request_type, name, response_time, response_length, request=None, **kwargs):
    if request and hasattr(request, "task_name"):
        new_name = f"{request.task_name}:{name}"
        events.request_success.fire(
            request_type=request_type,
            name=new_name,
            response_time=response_time,
            response_length=response_length,
            **kwargs
        )
        return False

@events.request_failure.add_listener
def adapt_failure_stats(request_type, name, response_time, exception, request=None, **kwargs):
    if request and hasattr(request, "task_name"):
        new_name = f"{request.task_name}:{name}"
        events.request_failure.fire(
            request_type=request_type,
            name=new_name,
            response_time=response_time,
            exception=exception,
            **kwargs
        )
        return False

适用场景：内部库直接使用requests发送请求，且允许传入Locust的client对象。

方法3：自定义TaskAwareSession（需内部库支持传参） #

如果内部库允许传入自定义的session对象，可以继承Locust的HttpSession，重写请求方法自动添加任务前缀。

from locust import HttpUser, task, HttpSession

class TaskAwareSession(HttpSession):
    def __init__(self, base_url, task_name=None, *args, **kwargs):
        self.task_name = task_name
        super().__init__(base_url, *args, **kwargs)

    def request(self, method, url, name=None, **kwargs):
        # 自动给请求统计名称加上任务前缀
        if name is None and self.task_name:
            name = f"{self.task_name}:{url}"
        return super().request(method, url, name=name, **kwargs)

class MyUser(HttpUser):
    @task
    def fetch_task(self):
        # 创建带任务名的session传入内部库
        session = TaskAwareSession(self.client.base_url, task_name="fetch")
        internal_lib.fetch(session)

    @task
    def update_task(self):
        session = TaskAwareSession(self.client.base_url, task_name="update")
        internal_lib.update(session)

适用场景：内部库支持接收自定义session参数，无需修改内部库的请求逻辑。

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