有 N 个 asyncio task，要求并发处理，并发量不超过 k，有什么优雅的写法？

[craftx]

[wliansheng]

不知道，楼主可以“抛砖引玉”？

[NessajCN]

啥叫并发量不超过 k ？
await asyncio.gather([小于 k 个 coro])
你是说这样？

[seers]

for 一下？

[alexsunxl]

k 数量大小的线程池？
完成一个 task 就回收到池子里。
思路应该是比较简单的吧

[BBCCBB]

asyncio 就是并发的?

再加个 Semaphore. 来控制同时最大请求数.

[centralpark]

asyncio.Semaphore 。不过这种问题适合问 ChatGPT 吧……

[Trim21]

Semaphore

[jonathanchoo]

您可以使用 asyncio 的 asyncio.gather() 方法来实现这个功能。您可以将所有的 asyncio task 放在一个 list 中，然后在 asyncio.gather() 方法中指定 concurrency 参数为 k ，即可实现并发量不超过 k 的并发处理。

以下是示例代码：

```python
import asyncio

async def task1():
# Your code here

async def task2():
# Your code here

# Put all tasks in a list
tasks = [task1(), task2(), ...]

async def main():
# Use asyncio.gather to run tasks concurrently with a maximum concurrency of k
await asyncio.gather(*tasks, return_exceptions=True, concurrency=k)

# Run the main function
asyncio.run(main())
```

[craftx]

@jonathanchoo
你好。在官方最新文档中，asyncio.gather 没有 concurrency 这个参数。
执行抛出异常：TypeError: gather() got an unexpected keyword argument 'concurrency'
https://docs.python.org/3/library/asyncio-task.html

[iorilu]

可以建立若干 worker ，N 个 worker 就控制并发 N 了把

我记得有这个模式

[zzl22100048]

创建一个
sem = asyncio.Semaphore(k)

async def task():
async with sem:
....

async def main():
await asyncio.gather(*[ task() for _ in range(n) ])

asyncio.run(main())



如果更喜欢 pipeline 语法，可以用 aiostream
from aiostream import pipe, stream
async def main():
await (
stream.iterate(range(n))
| pipe.map(task,task_limit=k)
)

[ruanimal]

@craftx 他用 gpt 生成的答案吧，可能是 gpt 编的

[photon006]

node.js 有个 promise 库 bluebird ，map()方法可以传一个参数 concurrency 控制并发量。

http://bluebirdjs.com/docs/api/promise.map.html

[zyxbcde]

@craftx 我是习惯把任务列表切成若干个不大于 k 的子列表，然后顺序去 gather 这些子列表，每跑完一个子列表打印个进度，好歹知道自己跑到哪了吧。
gather 里面从来就没有 concurrency 这个参数，这人用 gpt 生成个错误答案故意恶心人吧。

[craftx]

@zzl22100048 采用了 aiostream 的办法。谢谢