Python3.11 Asyncio：能否创建并发TaskGroup？取消后能否重启？

需求 #

我有一个Asyncio程序，需要实现：

• 任务0-2为安全任务，不受取消影响
• 任务3-5为可取消任务，触发异常时可被取消

尝试用Python3.11的TaskGroup管理可取消任务，同时用装饰器shield保护安全任务。

第一次尝试：单TaskGroup + shield装饰器 #

代码实现：

_task_list: List[asyncio.Task] = []

# 异常处理
try:
    async with self.task_group_safe as tg:
        for task in [
            self.task0(),
            self.task1(),
            self.task2(),
            self.task3(),
            self.task4(),
            self.task5(),
        ]:
            t = tg.create_task(task)
            _task_list.append(t)

except* AbortException as err:
    print(f"{err=}")
await asyncio.sleep(0)

问题 #

触发异常后，TaskGroup显示所有任务已被取消，尽管前3个任务被shield装饰且仍在运行：

for i, task in enumerate(_task_list):
    print(f"Task{i}: done={task.done()}, cancelled={task.cancelled()}")

输出结果：

Task0: done=True, cancelled=True
Task1: done=True, cancelled=True
Task2: done=True, cancelled=True
Task3: done=True, cancelled=True
Task4: done=True, cancelled=True
Task5: done=True, cancelled=False

第二次尝试：拆分到两个TaskGroup #

尝试将安全任务和可取消任务拆分到两个独立TaskGroup，无需给安全任务加shield：

_task_list: List[asyncio.Task] = []

tg1: List[Coroutine[Any, Any, Any]] = [
    self.task1(),
    self.task2(),
    self.task3(),
]
tg2: List[Coroutine[Any, Any, Any]] = [
    self.task4(),
    self.task5(),
    self.task6(),
]

async def create_task_group(
    task_group: asyncio.TaskGroup, tasks: List[Coroutine[Any, Any, Any]]
) -> None:
    # 异常处理
    try:
        async with task_group as tg:
            for task in tasks:
                t = tg.create_task(task)
                _task_list.append(t)
    except* AbortException as err:
        print(f"{err=}")
    await asyncio.sleep(0)

await create_task_group(self.task_group_1, tg1)
await create_task_group(self.task_group_2, tg2)

for i, task in enumerate(_task_list):
    print(f"Task{i}: done={task.done()}, cancelled={task.cancelled()}")

问题 #

程序卡在第一个create_task_group的await处，第二个TaskGroup无法创建。

第三次尝试：嵌套两个TaskGroup #

尝试在同一个上下文管理器中启动两个TaskGroup：

async def create_task_groups(
    task_group: asyncio.TaskGroup,
    tasks: List[Coroutine[Any, Any, Any]],
    task_group2: asyncio.TaskGroup,
    tasks2: List[Coroutine[Any, Any, Any]],
) -> None:
    # 异常处理
    try:
        async with task_group as tg, task_group2 as tg2:
            for task in tasks:
                t = tg.create_task(task)
                _task_list.append(t)
            for task in tasks2:
                t2 = tg2.create_task(task)
                _task_list.append(t2)
    except* AbortException as err:
        print(f"{err=}")
    await asyncio.sleep(0)

await create_task_groups(self.task_group_1, tg1, self.task_group_2, tg2)

问题 #

结果与第一次尝试一致，所有任务仍被标记为取消。

疑问 #

1. 能否让两个TaskGroup并发运行？
2. 若可以，取消第二个TaskGroup后能否重启它？

附：使用的shield装饰器代码 #

def _shielded(func: _AsyncFuncType) -> _AsyncFuncType:
    """
    让可等待方法始终不受取消影响
    """

    async def _shield(*args, **kwargs):
        return await asyncio.shield(func(*args, **kwargs))

    return _shield

其中_AsyncFuncType是Callable[..., Coroutine[Any, Any, Any]]的别名

内容的提问来源于stack exchange，提问作者OCopping