使用telegram.ext开发Telegram bot时如何为不同命令运行独立线程

注意telegram.ext不同版本的实现方案差异很大，先确认你用的版本再选择对应方案

适用v13.x及更早同步版本 #

方案1：使用框架内置run_async装饰器（推荐） #

telegram.ext 13+ 版本原生支持异步执行命令处理逻辑，只需要给处理函数加@run_async装饰器，框架会自动将命令处理逻辑丢到独立线程池运行，无需手动管理线程。
示例代码：

from telegram.ext import Updater, CommandHandler, run_async
import time

@run_async
def command_a(update, context):
    # 模拟耗时任务，不会阻塞其他命令处理
    time.sleep(10)
    update.message.reply_text("命令A执行完成")

@run_async
def command_b(update, context):
    time.sleep(5)
    update.message.reply_text("命令B执行完成")

def main():
    updater = Updater("你的机器人TOKEN", use_context=True)
    dp = updater.dispatcher
    # 注册命令处理器
    dp.add_handler(CommandHandler("a", command_a))
    dp.add_handler(CommandHandler("b", command_b))
    # 启动机器人
    updater.start_polling()
    updater.idle()

if __name__ == "__main__":
    main()

方案2：手动通过threading创建线程（自定义程度更高） #

如果需要自定义线程参数（比如线程名、daemon属性、线程组等），可以手动封装任务到线程执行：
示例代码：

from telegram.ext import Updater, CommandHandler
import threading
import time

# 命令A的实际任务逻辑
def task_a(update, context):
    time.sleep(10)
    update.message.reply_text("命令A执行完成")

def command_a(update, context):
    # 每次触发命令启动独立线程执行任务
    t = threading.Thread(target=task_a, args=(update, context), daemon=True)
    t.start()

# 命令B的实际任务逻辑
def task_b(update, context):
    time.sleep(5)
    update.message.reply_text("命令B执行完成")

def command_b(update, context):
    t = threading.Thread(target=task_b, args=(update, context), daemon=True)
    t.start()

def main():
    updater = Updater("你的机器人TOKEN", use_context=True)
    dp = updater.dispatcher
    dp.add_handler(CommandHandler("a", command_a))
    dp.add_handler(CommandHandler("b", command_b))
    updater.start_polling()
    updater.idle()

if __name__ == "__main__":
    main()

注意事项：

• 配置daemon=True可以保证主程序退出时自动终止所有未完成的任务线程，避免进程卡住无法退出
• 如果需要保证任务必须执行完成，不要设置daemon=True，需自行实现线程等待逻辑

适用v20.x及更高异步版本 #

v20之后的telegram.ext全面切换为asyncio异步模型，废弃了run_async装饰器，要实现命令独立运行不阻塞，可以通过asyncio.to_thread将同步耗时任务丢到独立线程执行，异步任务直接创建任务即可：
示例代码：

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

async def command_a(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # 同步耗时任务用asyncio.to_thread丢到独立线程
    await asyncio.to_thread(time.sleep, 10)
    await update.message.reply_text("命令A执行完成")

async def command_b(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await asyncio.to_thread(time.sleep, 5)
    await update.message.reply_text("命令B执行完成")

if __name__ == "__main__":
    app = ApplicationBuilder().token("你的机器人TOKEN").build()
    app.add_handler(CommandHandler("a", command_a))
    app.add_handler(CommandHandler("b", command_b))
    app.run_polling()

注意事项：

• 不要在异步处理函数中直接执行同步耗时操作，会阻塞整个事件循环，导致所有命令都无法响应，必须用asyncio.to_thread包裹同步耗时逻辑

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