如何在 Python/FastAPI 中监控事件循环中所有待执行的异步任务数量

如何在 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 服务吞吐量与响应延迟的正确路径。