V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
Learn Python the Hard Way
Python Sites
PyPI - Python Package Index
http://diveintopython.org/toc/index.html
Pocoo
值得关注的项目
PyPy
Celery
Jinja2
Read the Docs
gevent
pyenv
virtualenv
Stackless Python
Beautiful Soup
结巴中文分词
Green Unicorn
Sentry
Shovel
Pyflakes
pytest
Python 编程
pep8 Checker
Styles
PEP 8
Google Python Style Guide
Code Style from The Hitchhiker's Guide
craftx
V2EX  ›  Python

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

  •  
  •   craftx · 303 天前 · 1263 次点击
    这是一个创建于 303 天前的主题,其中的信息可能已经有所发展或是发生改变。
    15 条回复    2023-06-05 13:36:20 +08:00
    wliansheng
        1
    wliansheng  
       303 天前
    不知道,楼主可以“抛砖引玉”?
    NessajCN
        2
    NessajCN  
       303 天前
    啥叫并发量不超过 k ?
    await asyncio.gather([小于 k 个 coro])
    你是说这样?
    seers
        3
    seers  
       303 天前 via Android
    for 一下?
    alexsunxl
        4
    alexsunxl  
       303 天前
    k 数量大小的线程池?
    完成一个 task 就回收到池子里。
    思路应该是比较简单的吧
    BBCCBB
        5
    BBCCBB  
       303 天前
    asyncio 就是并发的?

    再加个 Semaphore. 来控制同时最大请求数.
    centralpark
        6
    centralpark  
       303 天前
    asyncio.Semaphore 。不过这种问题适合问 ChatGPT 吧……
    Trim21
        7
    Trim21  
       303 天前
    Semaphore
    jonathanchoo
        8
    jonathanchoo  
       303 天前
    您可以使用 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
        9
    craftx  
    OP
       303 天前
    @jonathanchoo
    你好。在官方最新文档中,asyncio.gather 没有 concurrency 这个参数。
    执行抛出异常:TypeError: gather() got an unexpected keyword argument 'concurrency'
    https://docs.python.org/3/library/asyncio-task.html
    iorilu
        10
    iorilu  
       303 天前
    可以建立若干 worker ,N 个 worker 就控制并发 N 了把

    我记得有这个模式
    zzl22100048
        11
    zzl22100048  
       302 天前
    创建一个
    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
        12
    ruanimal  
       302 天前
    @craftx 他用 gpt 生成的答案吧,可能是 gpt 编的
    photon006
        13
    photon006  
       302 天前
    node.js 有个 promise 库 bluebird ,map()方法可以传一个参数 concurrency 控制并发量。

    http://bluebirdjs.com/docs/api/promise.map.html
    zyxbcde
        14
    zyxbcde  
       302 天前
    @craftx 我是习惯把任务列表切成若干个不大于 k 的子列表,然后顺序去 gather 这些子列表,每跑完一个子列表打印个进度,好歹知道自己跑到哪了吧。
    gather 里面从来就没有 concurrency 这个参数,这人用 gpt 生成个错误答案故意恶心人吧。
    craftx
        15
    craftx  
    OP
       297 天前
    @zzl22100048 采用了 aiostream 的办法。谢谢
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   我们的愿景   ·   实用小工具   ·   1014 人在线   最高记录 6543   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 28ms · UTC 19:39 · PVG 03:39 · LAX 12:39 · JFK 15:39
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.