本文介绍在单进程内并行启动 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）。
以下是一个生产就绪的完整示例：
ImageMagick 批处理处理器
下载
使用 ImageMagick 的 convert 和 mogrify 命令进行批量图像处理，支持调整大小、裁剪、水印和格式转换。与 libvips 集成以实现高性能缩略图生成。
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 交互与外部系统集成能力。
