如何在python-telegram-bot(PTB)的另一个线程中运行异步函数

python-telegram-bot基于asyncio实现，所有操作默认在主线程的事件循环中执行，耗时的数据获取或计算会阻塞整个事件循环，导致机器人无法响应其他请求。以下是两种可行的解决方案：

方案一：调用API加速任务 #

如果耗时任务可以通过第三方服务或自建API完成，直接用异步HTTP库（如aiohttp）调用API获取结果，避免本地执行耗时操作。这种方式天然适配asyncio的非阻塞模型，不会阻塞主线程事件循环。

示例代码：

import aiohttp
from telegram import Update
from telegram.ext import ContextTypes

async def OptimizedSearch(context: ContextTypes.DEFAULT_TYPE) -> None:
    # 异步调用外部API完成耗时计算
    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://your-api-domain.com/heavy-calc",
            json={"task_param": "your_input_data"}
        ) as resp:
            if resp.status == 200:
                result = await resp.json()
            else:
                result = f"API调用失败，状态码: {resp.status}"
    
    await context.bot.send_message(
        chat_id=context.job.chat_id,
        text=str(result)
    )

async def run_script(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    job = context.job_queue.run_repeating(
        OptimizedSearch,
        interval=10*60,
        first=5,
        chat_id=chat_id,
        name="optimized_search"
    )

方案二：用后台线程运行异步任务 #

如果必须在本地执行耗时异步任务，可以通过asyncio.run_coroutine_threadsafe将任务提交到独立线程的事件循环中执行，避免阻塞主线程。

步骤1：启动后台线程与事件循环 #

在机器人初始化时（如main函数中），创建并启动一个后台线程，专门运行独立的asyncio事件循环：

import asyncio
import threading
from telegram.ext import ApplicationBuilder

# 创建后台事件循环并启动线程
background_loop = asyncio.new_event_loop()
background_thread = threading.Thread(
    target=lambda: (asyncio.set_event_loop(background_loop), background_loop.run_forever()),
    daemon=True
)
background_thread.start()

# 初始化机器人应用
def main():
    application = ApplicationBuilder().token("YOUR_BOT_TOKEN").build()
    # 注册命令处理器等逻辑
    application.run_polling()

if __name__ == "__main__":
    main()

步骤2：修改任务函数，提交耗时任务到后台循环 #

将原有的耗时逻辑抽离为独立协程，通过run_coroutine_threadsafe提交到后台循环，再用asyncio.wrap_future转换为可await的异步对象，避免阻塞主线程：

from telegram import Update
from telegram.ext import ContextTypes

# 耗时异步任务：数据获取与计算
async def heavy_task():
    # 模拟耗时操作，替换为实际业务逻辑
    await asyncio.sleep(30)
    return "耗时计算完成，结果为XXX"

async def OptimizedSearch(context: ContextTypes.DEFAULT_TYPE) -> None:
    # 将耗时协程提交到后台线程的事件循环
    concurrent_future = asyncio.run_coroutine_threadsafe(heavy_task(), background_loop)
    # 转换为asyncio可识别的Future，支持await非阻塞等待
    async_future = asyncio.wrap_future(concurrent_future)
    
    try:
        # 非阻塞等待任务结果，可设置超时
        result = await async_future
    except asyncio.TimeoutError:
        result = "任务执行超时，请稍后重试"
        concurrent_future.cancel()  # 取消后台任务
    except Exception as e:
        result = f"任务执行出错：{str(e)}"
    
    await context.bot.send_message(
        chat_id=context.job.chat_id,
        text=result
    )

async def run_script(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    job = context.job_queue.run_repeating(
        OptimizedSearch,
        interval=10*60,
        first=5,
        chat_id=chat_id,
        name="optimized_search"
    )

注意事项 #

• 后台线程设置为daemon=True，确保主程序退出时后台线程自动终止。
• 若耗时任务是同步代码（如CPU密集型计算），建议结合loop.run_in_executor使用线程池/进程池，而非直接用异步协程。
• 超时时间需根据实际任务耗时调整，避免无限制等待。

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