使用 asyncio.create_subprocess_exec 设置最大并发数

Question

我需要使用不同的输入运行一个程序大约 500 次。 我想使用

asyncio.create_subprocess_exec

并希望限制同时运行的进程数量，以免堵塞机器。 有没有办法设置并发级别？例如，我期望类似

AbstractEventLoop.set_max_tasks

。

Answer 1

正如 @AndrewSvetlov 建议，您可以使用

asyncio.Semaphore

来强制执行限制：

async def run_program(input):
    p = await asyncio.create_subprocess_exec(...)
    # ... communicate with the process ...
    p.terminate()
    return something_useful

async def run_throttled(input, sem):
    async with sem:
        result = await run_program(input)
    return result

LIMIT = 10

async def many_programs(inputs):
    sem = asyncio.Semaphore(LIMIT)
    results = await asyncio.gather(
        *[run_throttled(input, sem) for input in inputs])
    # ...

Answer 2

在需要调度足够的程序调用而立即将每个程序调用创建为协程的情况下，这个答案中的信号量支持的任务队列可能会很有用：

tasks = TaskQueue(NUM_PARALLEL_PROCESSES)
for input in MANY_INPUTS:
     await tasks.put(run_program(input))

tasks.join()