如何在 Python/FastAPI 中监控事件循环中所有待执行的异步任务数量
本文详细解析如何利用 asyncio.all_tasks() 函数获取当前事件循环中所有未完成（包括待执行与运行中）的 Task 对象，并通过完整的代码示例演示任务数量统计、状态日志记录与潜在阻塞堆栈分析，为 FastAPI 应用性能监控与问题排查提供实用工具。
在开发高性能 FastAPI 服务时，你是否曾遭遇接口响应延迟增加或系统资源消耗异常升高的情况？许多性能问题的早期征兆，往往源于异步任务的堆积与积压。首先需要明确一个核心概念：await 关键字本身并不会创建新的异步任务。它的作用仅是挂起当前协程的执行，等待被 await 的对象（例如 asyncio.sleep()、网络请求或数据库查询）返回结果。真正向事件循环提交独立、可调度执行单元的，是诸如 asyncio.create_task()、asyncio.ensure_future() 或 FastAPI 框架提供的 BackgroundTasks 这类显式调度操作。因此，我们关注的“待执行任务数量”，实质上指的是所有已提交至事件循环、但尚未结束（即 .done() 方法返回 False）的 Task 实例的总和，这涵盖了正在 CPU 上执行、暂停等待唤醒、阻塞于 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 参数。除非你显式管理多个事件循环，但这在标准的 FastAPI 应用架构中并不常见。
? 进阶监控：区分状态与诊断卡顿
仅了解任务总数有时不足以定位问题。若需识别哪些任务仍处于活跃状态，甚至探查可能发生阻塞的任务，则需要进行更深入的分析。以下代码展示了如何按状态过滤任务、记录详细日志以及输出调用堆栈信息：
立即学习“Python免费学习笔记（深入）”；
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() 的用法是构建异步应用可观测性能力的基础。它本身并不直接解决性能瓶颈，但能为你提供关键的第一手证据，精确揭示“事件循环当前正在处理哪些任务”。当你发现待执行任务数量持续增长且无法回落时，这便是一个明确的告警信号，提示你应深入分析对应协程的 I/O 效率、检查是否存在锁竞争，或排查是否混入了 CPU 密集型操作。这才是优化 FastAPI 服务吞吐量与响应延迟的正确路径。
