如何为Pyrogram开发的Telegram机器人进行负载测试？

一、避免自交互的核心思路：剥离业务与交互层 #

直接模拟MTProto更新触发机器人回复，必然导致自交互。正确的做法是把业务逻辑和Telegram交互代码完全分离，单独对核心业务逻辑做负载测试，再验证完整流程。

代码示例：抽离可测试的业务逻辑 #

将原来耦合在Pyrogram Handler里的逻辑，拆成独立的纯函数：

# 纯业务逻辑函数：只处理输入输出，不依赖Pyrogram上下文
def process_user_request(user_id: int, request_text: str) -> str:
    # 这里替换成你的复杂逻辑：数据库查询、计算、外部API调用等
    if request_text == "/start":
        return "欢迎使用机器人！"
    elif request_text.startswith("/calc"):
        try:
            num = int(request_text.split()[1])
            return f"计算结果：{num * 2}"
        except (IndexError, ValueError):
            return "请输入正确格式：/calc [数字]"
    # 其他功能逻辑...
    return "未知命令"

# Pyrogram Handler仅做消息转发与回复，无业务逻辑
@app.on_message(filters.command(["start", "calc"]))
async def handle_message(client, message):
    response = process_user_request(message.from_user.id, message.text)
    await message.reply(response)

对业务逻辑压测 #

用locust或pytest-benchmark直接压测process_user_request函数，无需涉及Telegram协议，完全避免自交互：

# 用locust压测业务逻辑的示例
from locust import User, task, between

class BusinessLogicUser(User):
    wait_time = between(0.1, 0.5)

    @task(3)
    def test_start_command(self):
        process_user_request(1001, "/start")

    @task(1)
    def test_calc_command(self):
        process_user_request(1002, "/calc 999")

二、真实场景负载测试：模拟多用户请求 #

如果要验证完整的机器人流程（含Telegram交互），需用独立的测试用户账号（而非机器人自身）模拟请求，避免自交互和Telegram限制。

代码示例：多账号并发请求脚本 #

import asyncio
from pyrogram import Client

# 测试账号列表：准备多个不同的Telegram账号
TEST_ACCOUNTS = [
    {"api_id": 123456, "api_hash": "your_api_hash_1", "phone": "+1234567890"},
    {"api_id": 123456, "api_hash": "your_api_hash_2", "phone": "+0987654321"},
    # 更多测试账号...
]
BOT_USERNAME = "your_bot_username"

async def simulate_user(account):
    # 初始化测试用户客户端
    client = Client(f"test_session_{account['phone']}", account["api_id"], account["api_hash"], phone_number=account["phone"])
    await client.start()
    
    # 模拟用户发送多种指令
    commands = ["/start", "/calc 100", "/calc 200"]
    for cmd in commands:
        await client.send_message(BOT_USERNAME, cmd)
        await asyncio.sleep(1)
    
    await client.stop()

async def run_load_test():
    # 并发执行多个用户的请求
    tasks = [simulate_user(acc) for acc in TEST_ACCOUNTS]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(run_load_test())

三、性能瓶颈定位 #

测试时同步监控机器人的资源使用和依赖服务状态：

import psutil
import time
import os

def monitor_bot_resources():
    bot_pid = os.getpid()  # 替换为机器人进程的PID
    process = psutil.Process(bot_pid)
    while True:
        cpu = process.cpu_percent(interval=1)
        memory = process.memory_info().rss / 1024 / 1024  # 转换为MB
        print(f"机器人CPU使用率: {cpu}% | 内存占用: {memory:.2f}MB")
        time.sleep(5)

# 在单独线程启动监控
import threading
threading.Thread(target=monitor_bot_resources, daemon=True).start()

四、关键注意事项 #

• 绝对不要用机器人账号给自己发消息，既会自交互，也可能触发Telegram反垃圾限制。
• 压测时逐步提升并发数，避免一次性发送大量请求导致账号被封禁。
• 如果机器人依赖数据库、外部API，需同步测试这些依赖服务的负载能力，避免瓶颈不在机器人本身。

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