python - 在 asyncio 中并行化 for 循环

Question

我目前有一个for循环如下

async def process(tasks, num):
      count = 0
      results = []
      for task in tasks:
           if count >= num:
               break
           result = await some_async_task(task)
           if result == 'foo':
               continue
           results.append(result)
           count+=1

我想知道我是否可以在这里使用 gather 或 wait_for 原语。但是我不确定如何在那里实现这些 if 逻辑？就像.. 如果计数> = num，我不想不必要地等待任务。如果有 20 个任务并且 num = 4，那么我不想运行所有 20 个任务。

Answer 1

这可以使用 aiostream library 轻松实现.这是一个工作示例:

import asyncio
from random import random
from aiostream import stream, pipe


async def some_async_task(i):
    await asyncio.sleep(random())
    return i if random() < 0.2 else None


async def process(task_args, n):
    return await (
        stream.iterate(task_args)
        | pipe.map(some_async_task, task_limit=n)
        | pipe.filter(bool)
        | pipe.take(n)
        | pipe.list()
    )


async def main():
    print(await process(task_args=range(100), n=10))


if __name__ == "__main__":
    asyncio.run(main())

程序打印前 10 个成功任务的列表:

[1, 8, 16, 18, 19, 37, 42, 43, 45, 47]

另请注意，您可以使用 task_limit 参数调整可同时运行的 some_async_task 的数量。

_{免责声明:我是项目维护者。}