lib模式下Locust停止运行时如何发送总请求统计报告到Telegram群组

核心原理 #

Lib模式下Locust的所有全局运行数据、统计结果都绑定在Environment实例上，你可以通过绑定environment.quit事件的回调函数实现需求：该事件会在整个测试完全结束、所有虚拟用户都退出后触发，此时可以直接读取到最终的全局汇总统计数据，不需要依赖每个用户的on_stop方法。

实现步骤 #

• 首先准备Telegram Bot的授权信息：提前在Telegram上申请Bot获取bot token，并将Bot加入目标群组获取对应的chat id
• 编写Telegram消息发送回调函数，函数入参为Locust的Environment实例，从实例的stats.total属性中提取全局统计指标组装报告内容，调用Telegram Bot接口发送消息
• 初始化Locust Environment实例后，将上述回调函数绑定到env.events.quit事件上
• 正常执行测试逻辑，测试停止时会自动触发回调发送报告

代码示例 #

from locust import HttpUser, task
from locust.env import Environment
import requests
import gevent

# Telegram配置项
TG_BOT_TOKEN = "替换为你的Telegram Bot Token"
TG_CHAT_ID = "替换为目标群组的Chat ID"

def send_telegram_report(env: Environment):
    # 获取全局汇总统计数据
    total_stats = env.stats.total
    # 组装报告内容，可按需扩展其他指标
    report = f"""
📊 Locust测试运行完成报告
=========================
总请求数：{total_stats.num_requests}
失败请求数：{total_stats.num_failures}
请求失败率：{total_stats.fail_ratio * 100:.2f}%
平均响应时间：{total_stats.avg_response_time:.2f}ms
95分位响应时间：{total_stats.get_response_time_percentile(0.95):.2f}ms
峰值RPS：{total_stats.total_rps:.2f}
    """.strip()
    # 调用Telegram接口发送消息
    requests.post(
        url=f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage",
        data={"chat_id": TG_CHAT_ID, "text": report}
    )

# 自定义测试用户类
class DemoUser(HttpUser):
    host = "https://example.com"
    @task
    def demo_request(self):
        self.client.get("/test/path")

if __name__ == "__main__":
    # 初始化Locust环境
    env = Environment(user_classes=[DemoUser])
    # 绑定测试停止事件的回调
    env.events.quit.add_listener(send_telegram_report)
    # 创建本地运行器（分布式场景下创建master runner即可）
    env.create_local_runner()
    # 启动测试：10个并发用户，孵化率1，运行30秒后停止
    env.runner.start(user_count=10, spawn_rate=1)
    gevent.sleep(30)
    env.runner.quit()

注意事项 #

• 如果使用分布式运行模式，仅需要在master节点的Environment实例上绑定事件回调即可，所有worker节点的统计数据会自动汇总到master节点的stats.total中，无需在worker侧添加逻辑
• 除了示例中提到的指标外，你可以从env.stats.total中获取最大/最小响应时间、各请求路径的单独统计、不同请求方法的统计等所有Locust默认统计的指标，按需组装报告即可

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