本文详解如何使用 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 实例。
? 深度诊断：日志化与栈追踪
单纯计数不够？可进一步打印任务详情或完整调用栈，快速识别“卡住”的协程：
立即学习“Python免费学习笔记（深入）”；
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
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)}响应示例：
{
"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() 使用） |
掌握 asyncio.all_tasks()，你就拥有了透视 FastAPI 异步心脏的“X光机”——无需第三方依赖，即可精准定位任务积压根源，让高并发下的性能调优从玄学回归工程实践。
