使用任务组响应

Question

在Python3.11中，建议使用TaskGroup生成任务而不是使用gather. 鉴于 Gather 还将返回协同例程的结果，那么使用 TaskGroup 的最佳方法是什么。

目前我有

async with TaskGroup() as tg:
      r1 = tg.create_task(foo())
      r2 = tg.create_task(bar())
res = [r1.result(), r2.result()]

是否有更简洁的方法可以达到相同的结果？

Answer 1

任务组的实现更多地是作为一种更清晰的方式来处理任务生命周期和异常处理，这在新的异常组返工的帮助下得到了极大的促进。我认为这是目前流行的误解，但这并不是所有用例的TaskGroup替代品。gather对于您不关心结果的情况（这似乎是我在新文档和教程中看到的唯一示例），感觉更加简洁。

当任务组完成后，用户仍然需要从已完成的协程中提取结果。res = [r1.result(), r2.result()]如果您立即需要值，那么您可以在完成后按原样编写TaskGroup。TaskGroup更简洁的语法可能是在完成后收集结果res = await asyncio.gather(r1, r2)（这将释放函数的执行，这可能是也可能不是所需的）。

使用TaskGroup和可能看起来多余gather，但它TaskGroup解决的目的与单独提供的目的不同gather，它允许等待具有强大安全保证的任务、围绕失败取消的逻辑以及异常分组。

可以扩展默认TaskGroup类以使此模式更容易。这是一个这样的想法，可以跟踪任务组中发出了哪些任务，并提供一个帮助程序来找出结果：

class GatheringTaskGroup(asyncio.TaskGroup):
    def __init__(self):
        super().__init__()
        self.__tasks = []

    def create_task(self, coro, *, name=None, context=None):
        task = super().create_task(coro, name=name, context=context)
        self.__tasks.append(task)
        return task

    def results(self):
        return [task.result() for task in self.__tasks]

async def foo(): return 1
async def bar(): return 2
async with GatheringTaskGroup() as tg:
    task1 = tg.create_task(foo())
    task2 = tg.create_task(bar())
print(tg.results())

[1, 2]