如何让python-telegram-bot与asyncio通知逻辑并发运行？

你的核心问题在于原代码中run_polling()是阻塞方法，会占据主线程导致后续的通知任务无法执行；同时通知任务仅执行一次而非循环检查更新。以下是基于asyncio的正确实现方式，完全适配python-telegram-bot的异步架构：

代码修改方案 #

1. 修复通知任务的循环逻辑 #

让通知任务持续运行，定期检查并推送新内容：

async def notify_about_new_tasks(self):
    while True:
        # 获取新任务列表（修正原代码的变量名笔误）
        new_tasks = await self.cheker.get_new_taks()
        for task in new_tasks:
            # 给所有订阅用户发送通知
            for subscription in self.subscriptions.subscriptions:
                await self.application.bot.send_message(
                    chat_id=subscription['chat_id'],
                    text=f"📢 发现新任务：{task}"  # 替换为实际任务内容
                )
        # 每隔100秒检查一次更新
        await asyncio.sleep(100)

2. 重构run方法为异步，整合两个任务 #

将机器人命令处理与通知任务放入同一个事件循环，实现并发执行：

async def run(self):
    # 添加命令处理器
    self.application.add_handler(CommandHandler("start", start))
    
    # 创建后台任务运行通知逻辑
    notify_task = asyncio.create_task(self.notify_about_new_tasks())
    
    try:
        # 启动机器人的轮询（异步非阻塞方式）
        await self.application.run_polling()
    finally:
        # 程序终止时清理通知任务
        notify_task.cancel()
        await notify_task

3. 主函数启动异步事件循环 #

if __name__ == "__main__":
    load_dotenv()
    client = Client()
    # 启动整个异步程序
    asyncio.run(client.run())

关键原理说明 #

• python-telegram-bot的Application.run_polling()本身是异步方法，直接await它不会阻塞事件循环，而是会让事件循环同时处理其他异步任务（比如通知任务）。
• asyncio.create_task()将通知任务注册到事件循环中，使其在后台持续运行，与机器人的命令处理逻辑并发执行，互不干扰。
• 避免使用多线程/多进程：因为python-telegram-bot的异步API并非线程安全，跨线程调用会引发潜在问题；而单事件循环下的异步任务是最安全高效的实现方式。

内容的提问来源于stack exchange，提问作者Not a cat Beaver