如何排查Python异步应用性能瓶颈：API延迟或asyncio就绪队列延迟？

核心思路 #

要区分两类延迟，关键是分别追踪任务在I/O等待阶段的耗时，以及任务就绪后到被事件循环调度执行的耗时，直接针对显式异步任务做精准监控。

具体监控方法 #

1. 给业务任务嵌入调度延迟追踪 #

直接在你关注的异步任务中添加时间戳，记录两个关键节点：

• I/O完成后任务就绪的时间点
• 任务实际开始执行的时间点

示例代码：

import asyncio
import time

async def your_api_task():
    # 记录I/O完成（任务就绪）的时间
    ready_time = time.monotonic()
    # 这里替换为实际的异步API/DB/Redis调用
    await asyncio.sleep(0)  # 模拟I/O完成后的就绪状态
    # 记录任务真正开始执行的时间
    exec_start_time = time.monotonic()
    
    # 计算调度延迟：就绪到执行的时间差
    schedule_delay = exec_start_time - ready_time
    print(f"当前任务调度延迟: {schedule_delay*1000:.2f}ms")
    
    # 执行实际业务逻辑，计算API本身耗时
    response = await asyncio.to_thread(lambda: "模拟API响应")  # 替换为真实异步客户端调用
    api_duration = time.monotonic() - exec_start_time
    print(f"API调用实际耗时: {api_duration*1000:.2f}ms")
    
    return response

这种方法无额外定时任务开销，直接针对目标任务，能精准拿到每个任务的调度延迟和业务I/O耗时。

2. 全局任务钩子自动追踪 #

通过自定义asyncio任务工厂，给所有异步任务自动注入调度延迟监控，无需修改业务代码：

import asyncio
import time

def tracked_task_factory(loop, coro):
    task = loop.create_task(coro)
    original_step = task._step
    
    async def tracked_step():
        ready_time = time.monotonic()
        # 执行原任务逻辑
        result = await original_step()
        exec_start_time = time.monotonic()
        # 计算并记录调度延迟
        schedule_delay = exec_start_time - ready_time
        # 可将数据上报到监控系统，比如打印或写入 metrics
        print(f"任务[{task.get_name()}]调度延迟: {schedule_delay*1000:.2f}ms")
        return result
    
    task._step = tracked_step
    return task

# 应用启动时设置
loop = asyncio.get_event_loop()
loop.set_task_factory(tracked_task_factory)

3. 统计分析定位瓶颈 #

收集所有任务的调度延迟和API/DB耗时的统计数据（平均值、中位数、95分位）：

• 如果调度延迟的分位数随并发量增长大幅上升，说明事件循环出现调度拥堵（比如存在CPU密集型任务阻塞循环，或任务数量过载）
• 如果API/DB耗时的分位数随并发量增长大幅上升，说明外部依赖服务本身存在并发瓶颈

额外排查点 #

• 检查是否有同步阻塞代码混入异步任务（比如误用同步客户端、调用time.sleep()而非asyncio.sleep()），这类代码会直接阻塞事件循环，导致所有任务调度延迟飙升
• 开启asyncio.debug()模式运行应用，排查是否有长时间占用事件循环的任务

内容的提问来源于stack exchange，提问作者Sumanth BBN