Python3.11中用TaskGroup替代gather获取协程结果的更简洁实现
使用TaskGroup获取任务结果的简洁写法
Python 3.11引入的TaskGroup是官方推荐替代gather的任务管理方案,你当前的实现是有效的,但可以通过以下方式让代码更简洁:
列表推导式批量处理任务
如果需要执行的任务数量较多或可批量生成,用列表推导式一次性创建任务,之后统一提取结果:
import asyncio async def foo(): return "foo result" async def bar(): return "bar result" async def main(): async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(coro()) for coro in [foo, bar]] res = [task.result() for task in tasks] print(res) asyncio.run(main())
固定任务数的解包写法
如果任务数量固定,直接通过解包赋值的方式创建任务,代码会更紧凑:
import asyncio async def foo(): return "foo result" async def bar(): return "bar result" async def main(): async with asyncio.TaskGroup() as tg: task_foo, task_bar = tg.create_task(foo()), tg.create_task(bar()) res = [task_foo.result(), task_bar.result()] print(res) asyncio.run(main())
需要注意的是,TaskGroup在退出async with上下文块时会自动等待所有内部任务完成,此时调用result()不会触发阻塞,这和gather自动返回结果的逻辑一致,同时保留了TaskGroup结构化任务管理的优势。
内容的提问来源于stack exchange,提问作者johng
相关产品推荐
相关产品推荐

