如何在Python中通过线程让Flask与Telegram Bot同时运行?
解决Flask与Telegram Bot同时运行的问题
你的线程方案可行,但需补充完整实现
你提到的多线程思路是有效的,但需要完善代码细节,把Telegram Bot的异步逻辑也整合进去:
1. 完善runner.py代码
要先导入Flask实例,初始化aiogram的Dispatcher,再用线程分别启动两个服务:
import threading import asyncio import logging from aiogram import Bot, Dispatcher from aiogram.enums.parse_mode import ParseMode import config from app import app # 导入Flask应用实例 # 初始化Telegram Bot的Dispatcher dp = Dispatcher() async def bot_main(): bot = Bot(token=config.BOT_TOKEN, parse_mode=ParseMode.HTML) await bot.delete_webhook(drop_pending_updates=True) await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()) def run_bot(): asyncio.run(bot_main()) if __name__ == "__main__": # 启动Flask服务,必须关闭自动重载(避免多线程冲突) flask_thread = threading.Thread(target=lambda: app.run(debug=False, use_reloader=False)) flask_thread.start() # 启动Telegram Bot run_bot()
2. 关键注意事项
- 关闭Flask自动重载:
use_reloader=False是必须的,否则自动重载会启动额外进程,打乱线程逻辑。 - 异步逻辑封装:Telegram Bot的异步代码要单独封装,用
asyncio.run()执行,避免和Flask的同步逻辑冲突。 - 数据库连接安全:Flask-SQLAlchemy默认会为每个线程创建独立数据库连接,一般无需额外配置;若出现连接异常,可检查
config.py中数据库连接池的参数设置。
生产环境推荐方案:异步Web服务器整合
如果是生产部署,更建议用支持异步的Web服务器(如Uvicorn),将Flask和Bot整合到同一个异步事件循环中,避免多线程潜在问题:
import asyncio from uvicorn import Server, Config from aiogram import Bot, Dispatcher from aiogram.enums.parse_mode import ParseMode import config from app import app dp = Dispatcher() async def bot_main(): bot = Bot(token=config.BOT_TOKEN, parse_mode=ParseMode.HTML) await bot.delete_webhook(drop_pending_updates=True) await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()) async def flask_main(): config = Config(app=app, host="0.0.0.0", port=5000) server = Server(config) await server.serve() async def main(): # 同时运行两个异步任务 await asyncio.gather(flask_main(), bot_main()) if __name__ == "__main__": asyncio.run(main())
内容的提问来源于stack exchange,提问作者WizAnd
相关产品推荐
相关产品推荐

