如何在 Python/FastAPI 中监控事件循环中的待处理异步任务？-Python教程-PHP中文网

如何在 Python/FastAPI 中监控事件循环中的待处理异步任务？

本文详解如何使用 asyncio.all_tasks() 实时获取并分析当前事件循环中所有未完成的异步任务，帮助定位 fastapi 应用性能瓶颈、任务积压或长时阻塞问题。

本文详解如何使用 asyncio.all_tasks() 实时获取并分析当前事件循环中所有未完成的异步任务，帮助定位 fastapi 应用性能瓶颈、任务积压或长时阻塞问题。

在基于 asyncio 的 Python Web 应用（如 FastAPI）中，事件循环是异步调度的核心。当请求量激增、I/O 延迟升高或协程中存在隐式同步阻塞（如未 await 的 CPU 密集操作、误用同步库）时，任务可能持续堆积在事件循环中，导致响应延迟、内存增长甚至服务不可用。此时，监控“待处理任务”（pending tasks）的数量与状态，是诊断性能问题的关键切入点。

✅ 正确获取所有待处理任务：asyncio.all_tasks()

Python 标准库提供了权威且轻量的方式——asyncio.all_tasks(loop=None)，它返回一个包含所有尚未完成（not yet done）的 Task 对象的 set，涵盖：

已调度但尚未开始执行的任务（pending）；
正在运行但被 await 暂停的任务（running/suspended）；
当前正在执行 all_tasks() 的任务本身（即调用者所在 Task）；
入口任务（如 asyncio.run() 包裹的主协程任务）。

import asyncio
import logging

# 在任意异步上下文中调用（例如 FastAPI 路由、中间件或健康检查端点）
async def get_pending_task_summary():
    tasks = asyncio.all_tasks()  # 获取当前运行 Loop 中所有未完成 Task
    pending_count = len(tasks)
    logging.info(f"当前事件循环中待处理任务总数: {pending_count}")

    # 可选：按状态分类统计
    running = [t for t in tasks if not t.done() and not t.cancelled()]
    pending = [t for t in tasks if t._state == "PENDING"]  # 注意：_state 是内部属性，仅作调试参考
    logging.debug(f"Running tasks: {len(running)}, Pending (internal): {len(pending)}")

    return {"total": pending_count, "running": len(running)}

⚠️ 注意：asyncio.all_tasks() 不区分“宏任务/微任务”（这是浏览器 JS 事件循环概念），而是面向 asyncio.Task 的统一视图——所有通过 asyncio.create_task()、asyncio.ensure_future() 或 await 自动调度的协程，最终都会表现为 Task 实例。

? 深度诊断：日志化与栈追踪

单纯计数不够？可进一步打印任务详情或完整调用栈，快速识别“卡住”的协程：

from fastapi import APIRouter
import asyncio
import traceback

router = APIRouter()

@router.get("/debug/tasks")
async def debug_tasks():
    tasks = asyncio.all_tasks()

    # 1. 日志化每个任务基本信息
    task_logs = []
    for task in tasks:
        task_info = {
            "name": task.get_name(),
            "done": task.done(),
            "cancelled": task.cancelled(),
            "exception": str(task.exception()) if task.done() and task.exception() else None,
        }
        task_logs.append(task_info)

    # 2. （可选）对运行中任务打印栈帧（谨慎使用，避免阻塞）
    for task in tasks:
        if not task.done() and not task.cancelled():
            # 使用 print_stack() 输出到 stderr，适合调试环境
            task.print_stack(limit=5)  # 仅显示最近 5 层调用

    return {"tasks": task_logs, "count": len(tasks)}

响应示例：

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

{
  "tasks": [
    {
      "name": "task-get-user",
      "done": false,
      "cancelled": false,
      "exception": null
    }
  ],
  "count": 42
}

? 提示：在生产环境建议将 print_stack() 替换为结构化日志（如 logging.debug(task.get_coro().__qualname__)），避免 I/O 阻塞或敏感信息泄露。

? 常见误区澄清

你提到：“每次 await 都创建一个待执行任务，慢则积压”。这不准确：

await 本身不创建新 Task，它只是暂停当前 Task 并让出控制权；
新 Task 仅在显式调用 create_task()、ensure_future()，或 FastAPI 自动包装路由协程时产生；
真正造成积压的是：
▪️ 高并发下未及时 await 的 I/O 操作（如数据库连接池耗尽）；
▪️ 协程内执行了同步阻塞操作（如 time.sleep()、requests.get()）；
▪️ 错误地将 CPU 密集型任务放入事件循环（应改用 loop.run_in_executor()）。

✅ 最佳实践建议

场景	推荐方案
实时监控	在 /health 或 Prometheus 指标端点中定期调用 len(asyncio.all_tasks())，设置告警阈值（如 > 100）
故障排查	结合 task.get_coro().__qualname__ 和 task.get_name() 快速定位异常模块
压力测试	在 Locust/Artillery 测试脚本中注入 debug_tasks() 调用，观察任务增长曲线
自动清理	对超时任务主动 task.cancel()（需配合 asyncio.wait_for() 使用）