本文介绍在单进程内并行启动 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 交互与外部系统集成能力。