在Django应用中运行python-telegram-bot避免阻塞服务器的方案咨询

问题描述 #

尝试在基于Django框架的Web应用中运行python-telegram-bot，按照Telegram库示例调用application.run_polling()方法时，该方法会阻塞Web服务器，想替换为更优方案却无从下手。

根据官方文档提示：

在将python-telegram-bot与其他基于asyncio的框架结合时，使用此方法并非最佳选择，因为它会阻塞事件循环直到收到停止信号。你可以手动调用以下方法来启动和关闭应用及更新器，维持事件循环运行并监听停止信号由你自行处理。

文档还列出了相关方法：

另请参阅：initialize(), start(), stop(), shutdown() telegram.ext.Updater.start_polling(), telegram.ext.Updater.stop(), run_webhook()

尝试理解这些方法后，仍无法实现python-telegram-bot在后台运行且不影响主服务器，需要此类场景的实现方案或详细步骤。

附当前使用的代码：

import asyncio

from django.apps import AppConfig
from telegram import Update
from telegram.ext import ContextTypes, Application, CommandHandler

from TaixTracking.configApp import ConfigApp


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Help!")


async def start_telegram():
    application = Application.builder().token(ConfigApp().get_value('telegram', 'token', '')).build()
    application.add_handler(CommandHandler("help", help_command))

    await application.initialize()
    await application.start()
    await application.updater.start_polling()
    print('Telegram load')


class CommunicationConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'communication'

    def ready(self):
        if not ConfigApp().is_telegram_launch():
            ConfigApp().set_telegram_launch(True)
            asyncio.run(start_telegram())

解决方案 #

方案1：使用线程单独运行Telegram Bot事件循环 #

Django的ready()方法为同步执行逻辑，直接调用asyncio.run()会阻塞主线程。可将Bot的异步逻辑放到单独线程中运行，避免影响Web服务器。

修改后代码：

import asyncio
import threading

from django.apps import AppConfig
from telegram import Update
from telegram.ext import ContextTypes, Application, CommandHandler

from TaixTracking.configApp import ConfigApp


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Help!")


async def start_telegram():
    application = Application.builder().token(ConfigApp().get_value('telegram', 'token', '')).build()
    application.add_handler(CommandHandler("help", help_command))

    await application.initialize()
    await application.start()
    # 启用Polling并允许丢弃未处理的更新
    await application.updater.start_polling(drop_pending_updates=True)
    # 保持事件循环运行直到Bot收到停止信号
    await application.idle()


def run_telegram_thread():
    # 为线程创建独立的asyncio事件循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(start_telegram())


class CommunicationConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'communication'

    def ready(self):
        if not ConfigApp().is_telegram_launch():
            ConfigApp().set_telegram_launch(True)
            # 启动守护线程运行Bot，Django退出时线程自动结束
            telegram_thread = threading.Thread(target=run_telegram_thread, daemon=True)
            telegram_thread.start()

关键改动：

• 为Bot创建独立线程，避免阻塞Django主线程
• 使用application.idle()维持事件循环持续运行
• 设置线程为守护线程，保证Django退出时线程自动终止

方案2：改用WebHook替代Polling（生产环境推荐） #

Polling需要持续请求Telegram服务器，WebHook则由Telegram主动推送更新到Django接口，更高效且不会占用线程持续运行。

步骤1：配置Telegram Bot的WebHook地址

新建Django管理命令完成WebHook配置：

# 在communication/management/commands下新建set_webhook.py
from django.core.management.base import BaseCommand
from telegram import Bot
from TaixTracking.configApp import ConfigApp

class Command(BaseCommand):
    help = '设置Telegram Bot的WebHook地址'

    def handle(self, *args, **options):
        bot_token = ConfigApp().get_value('telegram', 'token', '')
        # 替换为你的Django应用公网地址
        webhook_url = "https://your-domain.com/telegram/webhook/"
        bot = Bot(token=bot_token)
        bot.set_webhook(url=webhook_url)
        self.stdout.write(self.style.SUCCESS(f"WebHook已设置为：{webhook_url}"))

运行命令完成配置：python manage.py set_webhook

步骤2：创建接收WebHook的Django视图

# communication/views.py
from django.http import HttpResponse, HttpResponseBadRequest
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from asgiref.sync import sync_to_async
from TaixTracking.configApp import ConfigApp

# 全局单例Bot应用实例，避免重复初始化
bot_application = Application.builder().token(ConfigApp().get_value('telegram', 'token', '')).build()

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Help!")

# 注册命令处理器
bot_application.add_handler(CommandHandler("help", help_command))

async def process_telegram_update(update_data):
    update = Update.de_json(update_data, bot_application.bot)
    await bot_application.process_update(update)

def telegram_webhook(request):
    if request.method == "POST":
        update_data = request.body.decode('utf-8')
        # 异步处理Telegram更新，不阻塞Django请求
        sync_to_async(process_telegram_update)(update_data)
        return HttpResponse("OK")
    else:
        return HttpResponseBadRequest()

步骤3：配置路由

# communication/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('telegram/webhook/', views.telegram_webhook, name='telegram_webhook'),
]

方案3：ASGI模式下整合事件循环（适合异步Django） #

若使用Django 3.0+版本并搭配ASGI服务器（如Uvicorn），可直接将Bot任务加入Django的事件循环，无需额外线程：

import asyncio
from django.apps import AppConfig
from telegram import Update
from telegram.ext import ContextTypes, Application, CommandHandler
from TaixTracking.configApp import ConfigApp

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text("Help!")

async def start_telegram():
    application = Application.builder().token(ConfigApp().get_value('telegram', 'token', '')).build()
    application.add_handler(CommandHandler("help", help_command))
    await application.initialize()
    await application.start()
    await application.updater.start_polling(drop_pending_updates=True)
    await application.idle()

class CommunicationConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'communication'

    def ready(self):
        if not ConfigApp().is_telegram_launch():
            ConfigApp().set_telegram_launch(True)
            # 获取ASGI服务器的事件循环并添加Bot任务
            loop = asyncio.get_event_loop()
            loop.create_task(start_telegram())

注意：此方案仅适用于ASGI运行模式，WSGI模式下无法获取asyncio事件循环。

关键注意事项 #

• 生产环境优先选择WebHook方案，Polling仅适合开发调试场景
• 确保Bot的Application实例为全局单例，避免重复初始化
• 使用线程方案时，需通过配置或缓存标记Bot是否已启动，防止Django重启时重复创建线程

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