为什么创建大量极小的Python asyncio任务时性能会下降？

当创建10万个仅包含await asyncio.sleep(0)的极小异步任务时，性能下降主要来自三个核心原因：

• Task对象的固有开销：每个asyncio.Task都需要维护状态、回调、上下文等信息，10万个这样的实例会占用大量内存，同时事件循环调度时需要逐个遍历处理这些任务，带来显著的调度成本。
• 频繁的任务切换：await asyncio.sleep(0)的作用是主动触发一次任务切换，让事件循环调度其他就绪任务。10万次这样的切换会让事件循环在调度逻辑上消耗大量CPU时间，而非执行实际业务代码。
• gather的管理开销：asyncio.gather需要跟踪所有任务的完成状态，维护一个包含10万个任务的列表会带来额外的内存和计算开销。

优化方案 #

1. 批量合并小任务

把多个极小的逻辑合并到同一个任务中执行，直接减少Task的创建数量，从根源上降低调度和内存开销。比如将每1000个小逻辑打包成一个任务：

import asyncio
import time

async def batch_task(count):
    for _ in range(count):
        # 直接执行原tiny_task的逻辑，无需触发任务切换
        pass  # 原逻辑是await asyncio.sleep(0)，无实际等待时可直接跳过

async def main():
    batch_size = 1000
    total_tasks = 100000
    tasks = [asyncio.create_task(batch_task(batch_size)) for _ in range(total_tasks // batch_size)]
    await asyncio.gather(*tasks)

start = time.time()
asyncio.run(main())
print(time.time() - start)

如果实际小任务确实需要触发切换，也可以在批量任务中周期性触发切换，平衡任务数量和调度频率：

async def batch_task(count):
    for i in range(count):
        # 执行你的小逻辑
        if i % 100 == 0:
            await asyncio.sleep(0)  # 每100次逻辑触发一次切换

2. 避免不必要的异步包装

如果小任务没有实际的IO等待或异步操作，完全不需要包装成asyncio.Task，直接在主协程中同步执行即可，性能会大幅提升：

import asyncio
import time

def tiny_task():
    pass  # 无异步操作的逻辑直接同步执行

async def main():
    for _ in range(100000):
        tiny_task()

start = time.time()
asyncio.run(main())
print(time.time() - start)

3. 控制并发任务数量

如果必须保留独立任务，可以通过限制并发数来减少事件循环的调度压力，比如用信号量控制同时运行的任务数：

import asyncio
import time

async def tiny_task(sem):
    async with sem:
        await asyncio.sleep(0)

async def main():
    sem = asyncio.Semaphore(1000)  # 限制同时运行1000个任务
    tasks = [asyncio.create_task(tiny_task(sem)) for _ in range(100000)]
    await asyncio.gather(*tasks)

start = time.time()
asyncio.run(main())
print(time.time() - start)

测试对比 #

以原代码为例，在普通PC上运行可能需要几秒甚至十几秒；而批量处理的版本通常能把时间压缩到几十毫秒以内，性能提升非常显著。

内容的提问来源于stack exchange，提问作者Indrajith Bandara