如何在Locust中为非HTTP调用手动标记请求成功或失败？

你可以直接使用Locust官方支持的self.environment.events.request.fire方法来手动触发请求的成功或失败统计，无需依赖虚拟HTTP请求这种hack手段。这个方法专门用于记录自定义请求（比如数据库调用、SDK请求等非HTTP场景）的指标，会在Locust UI中正常展示成功率、响应时间等数据。

实现示例 #

修改你的read_item方法，添加耗时计算和事件触发逻辑：

import time
from locust import User, task

class YourUserClass(User):
    @task(100)
    def read_item(self) -> None:
        start_time = time.time()
        request_name = "DB-read_data"  # 自定义请求名称，方便UI识别
        try:
            # 执行数据库SDK调用
            database_to_read, attribute_to_read, key_to_read = self.sdk[database_to_read].read_data(
                conditions={condition_id: key_to_read}
            )
            # 计算请求耗时（转毫秒）
            response_time = int((time.time() - start_time) * 1000)
            # 触发成功事件
            self.environment.events.request.fire(
                request_type="DB",  # 自定义请求类型，比如"DB"表示数据库请求
                name=request_name,
                response_time=response_time,
                response_length=0,  # 非HTTP请求无响应长度时设为0
                exception=None
            )
        except Exception as e:
            response_time = int((time.time() - start_time) * 1000)
            # 触发失败事件
            self.environment.events.request.fire(
                request_type="DB",
                name=request_name,
                response_time=response_time,
                response_length=0,
                exception=e
            )

参数说明 #

• request_type：自定义请求类型（比如"DB"、"SDK"），用于在UI中区分不同类型的请求
• name：请求名称，建议设置为有意义的标识（如DB-read_data），方便统计分析
• response_time：请求耗时（毫秒），需要手动计算
• response_length：响应长度，非HTTP请求可设为0
• exception：失败时传入捕获的异常对象，成功时设为None

这种方式完全符合Locust的设计规范，替代了你之前用虚拟HTTP请求的临时方案，统计数据会准确展示在Locust的仪表盘和报告中。

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