本文详解如何使用 asyncio.all_tasks() 获取当前事件循环中所有未完成（pending 或 running）的 task 对象，并通过实际代码演示任务统计、日志记录与堆栈分析，助力 fastapi 应用性能排查。
本文详解如何使用 asyncio.all_tasks() 获取当前事件循环中所有未完成（pending 或 running）的 task 对象，并通过实际代码演示任务统计、日志记录与堆栈分析，助力 fastapi 应用性能排查。
在构建高并发 FastAPI 服务时，异步任务的积压往往是响应延迟或资源耗尽的早期信号。但需明确：await 本身不会自动创建新任务——它只是挂起当前协程，等待被 await 的可等待对象（如 asyncio.sleep()、HTTP 请求、数据库查询）完成；真正生成独立可调度任务的是 asyncio.create_task()、asyncio.ensure_future() 或 FastAPI 内部的 BackgroundTasks 等显式调度行为。因此，“待执行任务数”反映的是已提交至事件循环、尚未完成（.done() 为 False）的所有 Task 实例总数，包括正在运行、已暂停、等待 I/O 或处于队列中的任务。
✅ 获取待执行任务数量（核心方法）
最直接的方式是调用 asyncio.all_tasks()，它返回一个 set，包含当前运行事件循环中所有未完成的 Task 对象：
import asyncio
# 在任意协程内（如 FastAPI 路由处理函数中）
async def monitor_tasks():
pending_tasks = asyncio.all_tasks()
count = len(pending_tasks)
print(f"当前待执行/运行中的任务总数: {count}")
return count
⚠️ 注意：asyncio.all_tasks() 默认使用 asyncio.get_running_loop() 获取当前事件循环，无需手动传入 loop 参数（除非你显式管理多个 loop，这在标准 FastAPI 场景中极少见）。
? 进阶监控：区分状态与诊断卡顿
仅知总数不够？可进一步过滤活跃任务（非 done）、打印详情或导出堆栈：
import asyncio
import logging
logging.basicConfig(level=logging.DEBUG)
async def diagnose_task_backlog():
tasks = asyncio.all_tasks()
# 1. 统计：总任务数 vs 活跃任务数（未完成）
total = len(tasks)
active = len([t for t in tasks if not t.done()])
print(f"[监控] 总任务: {total}, 活跃中: {active}")
# 2. 日志所有任务基本信息（推荐用于 Prometheus / Grafana 集成）
for task in tasks:
logging.debug(
f"Task {task.get_name() or 'unnamed'} | "
f"State: {'DONE' if task.done() else 'PENDING'} | "
f"Coro: {task.get_coro().__qualname__ if task.get_coro() else 'N/A'}"
)
# 3. （谨慎使用）打印卡住任务的完整调用栈（定位阻塞点）
for task in tasks:
if not task.done() and task._coro.cr_await is None: # 粗略判断“疑似卡住”
print(f"\n⚠️ 可能卡住的任务 {task.get_name()} 堆栈:")
task.print_stack(limit=10)
# 在 FastAPI 中作为依赖或中间件调用示例
from fastapi import Depends, APIRouter
router = APIRouter()
@router.get("/health/tasks")
async def get_task_status():
await diagnose_task_backlog()
return {"status": "ok"}
? 关键注意事项
- all_tasks() 包含自身：调用该函数的协程本身也被包装为一个 Task（例如 FastAPI 路由处理函数），因此总数恒 ≥ 1。
- 不包含普通协程（coroutine）：只有被 create_task()、ensure_future() 或框架（如 FastAPI）显式调度的才计入；裸 await coro() 不新增任务。
- 生命周期敏感：all_tasks() 返回的是瞬时快照。任务可能在你遍历时已完成，建议在关键路径（如请求入口/出口）或定时健康检查中调用。
-
FastAPI 生产环境建议：
- 避免高频调用（如每个请求都 print_stack），可用 logging.debug + 条件采样；
- 结合 asyncio.current_task().get_name() 为任务命名，提升可读性：
task = asyncio.create_task(some_coro(), name="db-query-user-profile")
- 替代方案（更轻量）：若仅需计数，len(asyncio.all_tasks()) 是最低开销方式；避免无谓遍历。
掌握 asyncio.all_tasks() 是异步应用可观测性的基石。它不解决性能瓶颈本身，但为你提供了精准定位“谁在占用事件循环”的第一手证据——当 pending_tasks 持续增长且无法回落时，即可锁定对应协程深入分析 I/O 效率、锁竞争或 CPU 密集型操作，从而真正优化 FastAPI 服务的吞吐与延迟表现。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
