Python如何限制异步并行awaitable任务最大数量避免OOM

方案1：标准库原生实现（零依赖，推荐首选） #

用asyncio.Semaphore（信号量）搭配asyncio.as_completed即可实现无批次等待的动态限流，同时严格控制并发任务数不超过你设定的阈值：

import asyncio
from random import sample

# 替换为你需要的最大并行数X
MAX_CONCURRENT_TASKS = 10

async def bounded_async_process(semaphore, target):
    # 信号量会限制同时进入这段代码的协程数量
    async with semaphore:
        return await self.slowAsyncProcessing(target)

async def process_tasks():
    sample_size = self.targets / TEST_SAMPLE_RATIO
    targets = sample(self.targets, int(sample_size))
    semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS)
    # 生成所有待执行任务
    tasks = [bounded_async_process(semaphore, t) for t in targets]
    # 任务按完成顺序返回，无需等待整批完成
    for task in asyncio.as_completed(tasks):
        result = await task
        fastNonAsyncProcessingOfTheResult(result)

# 运行协程
asyncio.run(process_tasks())

这个实现完全没有你原来批次处理的短板效应，只要有一个任务完成释放信号量，下一个等待的任务就会立刻启动，全程保持最多X个任务并行执行，资源利用率拉满。

方案2：第三方库极简实现 #

如果希望进一步简化代码，可以用成熟的异步限流库aiometer，仅需几行代码即可实现同等效果：

1. 先安装依赖：pip install aiometer
2. 代码示例：

import asyncio
import aiometer
from random import sample

async def process_tasks():
    sample_size = self.targets / TEST_SAMPLE_RATIO
    targets = sample(self.targets, int(sample_size))
    # 直接指定max_at_once参数控制最大并行数
    async with aiometer.amap(
        self.slowAsyncProcessing,
        targets,
        max_at_once=10,
    ) as results:
        async for result in results:
            fastNonAsyncProcessingOfTheResult(result)

asyncio.run(process_tasks())

方案3：超大量任务适配方案 #

如果你的任务量级达到十万甚至百万级别，一次性生成所有任务可能占用过多内存，可以用原生asyncio.Queue实现生产者消费者模式，内存占用长期稳定：

import asyncio
from random import sample

MAX_CONCURRENT_TASKS = 10

async def task_worker(queue):
    while True:
        target = await queue.get()
        try:
            result = await self.slowAsyncProcessing(target)
            fastNonAsyncProcessingOfTheResult(result)
        finally:
            queue.task_done()

async def process_tasks():
    sample_size = self.targets / TEST_SAMPLE_RATIO
    targets = sample(self.targets, int(sample_size))
    queue = asyncio.Queue(maxsize=MAX_CONCURRENT_TASKS)
    # 启动固定数量的消费协程
    workers = [asyncio.create_task(task_worker(queue)) for _ in range(MAX_CONCURRENT_TASKS)]
    # 生产者往队列投放任务，队列满时自动阻塞
    for t in targets:
        await queue.put(t)
    # 等待所有任务处理完成
    await queue.join()
    # 销毁worker协程
    for w in workers:
        w.cancel()

asyncio.run(process_tasks())

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