如何在python-telegram-bot v20+中异步运行Telegram机器人轮询且不阻塞主应用

哥们，我之前在python-telegram-bot v20+里也踩过这个坑！这个版本完全切换到asyncio异步架构了，和老版本的处理逻辑不一样，之前瞎用线程或者直接asyncio.create_task()踩了一堆循环冲突的错误，后来摸清楚了正确的玩法，给你两个靠谱的方案：

方案一：主应用本身是异步的情况（推荐） #

如果你的主应用逻辑也是异步的，直接用asyncio.gather()把机器人轮询和主应用任务放在同一个事件循环里调度就行，这样两者都不会互相阻塞。

示例代码：

import asyncio
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# 机器人的命令处理器
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('Hello!')

# 你的主应用异步逻辑，比如定时任务、其他异步操作
async def main_app_task():
    while True:
        print("主应用在正常运行...")
        await asyncio.sleep(5)

async def main():
    # 初始化机器人应用
    application = ApplicationBuilder().token("你的机器人Token").build()
    application.add_handler(CommandHandler("start", start))

    # 同时启动轮询和主应用任务
    await asyncio.gather(
        application.run_polling(),
        main_app_task()
    )

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

原理很简单：application.run_polling()本身就是个异步协程，asyncio.gather()会把它和你的主应用协程一起扔进事件循环，由asyncio自动调度执行，完全不会出现阻塞的问题。

方案二：主应用是同步的情况 #

如果你的主应用是同步代码（比如传统的脚本、同步Web框架），那得把机器人的异步逻辑放到单独的线程里，而且要注意在子线程里单独创建事件循环，不能和主线程的循环冲突。

示例代码：

import asyncio
import threading
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes

# 机器人命令处理器
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('Hello!')

# 子线程中运行机器人的逻辑
def run_bot_in_thread():
    # 必须在子线程里新建一个事件循环，不然会和主线程的循环冲突
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    application = ApplicationBuilder().token("你的机器人Token").build()
    application.add_handler(CommandHandler("start", start))
    # 在子线程的循环里运行轮询
    loop.run_until_complete(application.run_polling())

# 你的同步主应用逻辑
def main_sync_app():
    while True:
        print("同步主应用在正常运行...")
        # 模拟主应用的同步工作
        import time
        time.sleep(3)

if __name__ == "__main__":
    # 把机器人启动在后台线程，设置daemon=True让主应用退出时线程自动结束
    bot_thread = threading.Thread(target=run_bot_in_thread, daemon=True)
    bot_thread.start()

    # 运行你的同步主应用
    main_sync_app()

为啥你之前会踩坑？ #

你遇到的RuntimeError: This event loop is already running或者Cannot close a running event loop，本质都是事件循环的冲突：

• 如果你在已经运行的事件循环里直接调用asyncio.create_task(application.run_polling())，但又没正确把它加入到当前循环的调度里，或者重复启动循环；
• 用线程但没在子线程新建事件循环，直接用主线程的循环，导致两个线程抢同一个循环资源。

额外注意点 #

• 别多次调用asyncio.run()，它每次都会新建一个事件循环，运行完就关闭，重复调用必然出问题；
• 如果用的是FastAPI这类异步Web框架，可以把机器人轮询加到框架的后台任务里，逻辑和方案一类似；
• 要是用Flask、Django这种同步Web框架，直接用方案二的线程方式就行，别在框架的请求上下文里碰机器人的异步逻辑。

备注：内容来源于stack exchange，提问作者Uccio Maselli