如何实现Aiogram版Telegram Bot与Flask服务同时运行？

你之前的两种实现跑不通的核心原因很明确：

• 多线程版本的事件循环初始化顺序完全错误，异步启动前没有给对应线程绑定独立事件循环，且同步的Flask服务和aiogram的异步事件循环没有做线程隔离
• on_startup版本直接把同步阻塞的app.run()当成协程await，直接堵死了aiogram的整个事件循环，导致bot逻辑拿不到执行权

以下是三种经过验证的可落地方案，按需选择即可：

方案1：修正多线程逻辑，保留现有Flask+Polling架构 #

这个方案改动最小，不需要替换现有技术栈，核心是做好线程和事件循环的隔离，要点如下：

• 给aiogram开辟独立的工作线程，线程启动第一时间就为该线程创建并绑定专属的asyncio事件循环，再启动polling
• Flask服务跑在独立线程中，二者完全通过线程安全的方式交互，禁止跨线程直接调用/await aiogram的协程方法
• 如果Flask的接口逻辑需要触发bot发消息、执行操作，统一用asyncio.run_coroutine_threadsafe()把协程任务提交到bot线程的专属事件循环中执行

修正后的核心代码参考：

import threading
from time import sleep
from flask import Flask, request
from flask_restful import Api
import config
import asyncio
from aiogram import Bot, Dispatcher, executor, types
from asyncio import new_event_loop, set_event_loop

app = Flask(__name__)
api = Api(app)
bot = Bot(token=config.bot_token)
dp = Dispatcher(bot)
bot_loop = None  # 全局存储bot线程的事件循环，供Flask侧调用

@dp.message_handler(commands="start")
async def message_start(message: types.Message):
    await message.answer("HI")

@app.route('/test', methods=['POST'])
def get_test():
    data = request.get_json()
    # 线程安全调用bot方法示例
    # asyncio.run_coroutine_threadsafe(
    #     bot.send_message(chat_id=data.get("chat_id"), text="收到请求"),
    #     loop=bot_loop
    # )
    print(data)
    return ("ok")

def app_run():
    while True:
        try:
            # 生产环境不要用Flask自带的开发服务器，替换为gunicorn等WSGI服务器
            app.run(host="0.0.0.0", port=5000, debug=False, use_reloader=False)
        except Exception as ex:
            print(ex)

def bot_polling():
    global bot_loop
    # 线程启动第一步：初始化当前线程的专属事件循环
    bot_loop = new_event_loop()
    set_event_loop(bot_loop)
    try:
        executor.start_polling(dp, skip_updates=True, loop=bot_loop)
    except Exception as ex:
        print(ex)

if __name__ == "__main__":
    bot_thread = threading.Thread(target=bot_polling, daemon=True)
    bot_thread.start()

    app_thread = threading.Thread(target=app_run, daemon=True)
    app_thread.start()

    while True:
        try:
            sleep(1)
        except KeyboardInterrupt:
            break

方案2：替换Flask为ASGI框架，和aiogram同事件循环运行 #

多线程天然存在线程安全问题，更优雅的方式是全栈用异步实现，彻底抛弃多线程：

• 把同步的Flask换成和Flask语法几乎一致的异步ASGI框架Quart（或者FastAPI），原有Flask路由代码改动量极小，只需要给视图函数加async关键字即可
• 用uvicorn作为ASGI服务器，把aiogram和ASGI API服务挂载到同一个事件循环中运行，两边可以直接互相await调用，没有任何线程冲突
• 这个方案既可以搭配polling模式，也可以搭配webhook模式，代码结构最干净，维护成本最低

方案3：用aiogram Webhook模式完全替代Flask #

这个方案完全可行，但是有明确的适用前提：

• 你的服务必须有公网可访问的HTTPS地址，Telegram要求webhook地址必须为HTTPS协议，且仅支持80/443/88/8443端口
• 满足前提的情况下，你不需要单独部署Flask，aiogram内置的web服务既可以接收Telegram推送的消息更新，也支持注册自定义路由处理你原来的第三方POST请求，相当于把API服务和Bot服务合二为一，性能比polling模式更好，不需要持续轮询Telegram服务器
• 如果你的服务部署在内网、没有公网HTTPS域名，这个方案就没法用，还是选前两种polling相关的方案

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