AIogram3与aiohttp共存运行教程

大家好，我们又见面了啊~本文《AIogram 3 与 aiohttp 同时运行教程》的内容中将会涉及到等等。如果你正在学习文章相关知识，欢迎关注我，以后会给大家带来更多文章相关文章，希望我们能一起进步！下面就开始本文的正式内容~

本文介绍在单进程内并行启动 aiogram 3 轮询模式机器人和 aiohttp 异步 Web 服务器的正确方法，避免事件循环冲突，并提供可维护、符合官方实践的结构化实现方案。

在构建需要“双向通信”的 Telegram Bot 应用时（例如：Bot 主动推送消息给用户，同时接收外部系统通过 HTTP 发起的通知），常需将 aiogram 3 的轮询机器人与 aiohttp Web 服务共存于同一 Python 进程中。但直接调用 dp.start_polling() 和 web.run_app() 会引发冲突——因为二者都试图接管事件循环，且 web.run_app() 是阻塞式入口函数，不可与 asyncio.run() 或手动 loop.run_until_complete() 混用。

✅ 正确做法是：让 aiohttp 成为“主应用”，通过 cleanup_ctx 上下文管理器异步启动并生命周期托管 aiogram 任务。这种方式更健壮、可扩展，且与 aiohttp 官方推荐的复杂应用架构一致（见 aiohttp 文档 - Complex Applications）。

以下是一个生产就绪的完整示例：

import asyncio
import logging
from contextlib import suppress
from aiogram import Bot, Dispatcher
from aiohttp import web

# 配置日志（建议启用）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

TOKEN = "YOUR_BOT_TOKEN_HERE"

# 1. 定义 bot 启动逻辑（作为独立协程）
async def start_aiogram_bot():
    bot = Bot(TOKEN, parse_mode="HTML")
    dp = Dispatcher()

    # ✅ 注册你的处理器（如 /start、消息处理等）
    # @dp.message(Command("start"))
    # async def cmd_start(message: Message):
    #     await message.answer("Hello from aiogram!")

    logger.info("Starting aiogram bot in polling mode...")
    await dp.start_polling(bot)

# 2. 定义 aiohttp Web 服务（含 bot 生命周期管理）
async def init_web_app() -> web.Application:
    app = web.Application()

    # ✅ 使用 cleanup_ctx 确保 bot 与 web server 共享同一事件循环
    # 并在应用关闭时优雅终止 bot 任务
    async def run_bot_task(_app):
        task = asyncio.create_task(start_aiogram_bot())

        yield  # 应用运行期间保持 bot 活跃

        # 清理阶段：取消 bot 任务
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task  # 等待任务完全退出，抛出异常（如有）

    app.cleanup_ctx.append(run_bot_task)

    # ✅ 添加你的 HTTP 处理器（例如接收外部通知）
    async def handle_notify(request):
        try:
            data = await request.json()
            user_id = data.get("user_id")
            text = data.get("text", "Notification received.")

            # 注意：此处需访问 bot 实例 —— 推荐使用全局变量或依赖注入
            # 实际项目中建议通过 app['bot'] 注入（见进阶说明）
            # 示例暂略 bot 发送逻辑（需确保 bot 实例可访问）

            return web.json_response({"status": "ok", "received": True})
        except Exception as e:
            logger.error(f"Failed to handle notify: {e}")
            return web.json_response({"error": str(e)}, status=400)

    app.router.add_post("/notify", handle_notify)

    return app

# 3. 启动入口（标准 aiohttp 方式）
if __name__ == "__main__":
    web_app = init_web_app()
    web.run_app(web_app, host="127.0.0.1", port=5555)

? 关键要点说明：

• web.run_app() 是 aiohttp 的顶层运行器，它内部调用 asyncio.run()，因此不应再手动获取/操作事件循环（如 get_event_loop()、create_task() 等）；
• cleanup_ctx 是 aiohttp 提供的生命周期钩子：yield 前执行初始化（启动 bot），yield 后执行清理（取消 bot）；
• bot 任务被包裹在 asyncio.create_task() 中，保证其在后台持续运行，不阻塞 Web 路由响应；
• 所有异步资源（如数据库连接、HTTP 客户端）均可通过 app['key'] 注入并在 cleanup_ctx 中统一释放，提升可维护性。

⚠️ 注意事项：

• 若需在 Web handler 中调用 bot.send_message()，请将 Bot 实例存储于 app['bot']（在 run_bot_task 初始化后赋值），而非创建新实例；
• 避免在 start_aiogram_bot() 内部直接使用 await dp.start_polling() 后继续执行其他逻辑——start_polling() 是长期运行协程，会阻塞后续代码；cleanup_ctx 的 yield 正是为此设计；
• 开发调试时可添加 --reload 支持（需 aiohttp-devtools），但注意热重载可能干扰 bot 状态，生产环境应禁用。

通过该结构，你获得了一个统一调度、生命周期可控、易于监控和部署的混合异步服务，兼顾 Telegram 交互与外部系统集成能力。

今天关于《AIogram3与aiohttp共存运行教程》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！