TaskGroup不能存为类属性或实例属性，因为它是单次作用域的一次性上下文管理器，async with退出后即失效，再调用create_task会抛RuntimeError: Task group is closed。
TaskGroup 在 Python 3.11 中不能直接用于类方法内部的并发任务管理——它必须在 async 函数中被 async with 语句正确进入，且不能跨协程生命周期持有或复用。
为什么不能把 TaskGroup 实例存成类属性或实例属性？
TaskGroup 不是可重入、不可暂存的状态容器。它的设计目标是一次性作用域：创建 → 进入 → 启动子任务 → 退出时自动等待并传播异常。一旦 async with task_group: 块结束，task_group 就失效，再调用 task_group.create_task(...) 会抛出 RuntimeError: Task group is closed。
常见错误写法：
class Worker:
def __init__(self):
self.tg = asyncio.TaskGroup() # ❌ 错误：未在 async 上下文中初始化
<pre class="brush:php;toolbar:false;"><code>async def run_all(self):
async with self.tg: # ❌ 即使这里进了，退出后 self.tg 就废了
self.tg.create_task(self.fetch_a())
self.tg.create_task(self.fetch_b())
立即学习“Python免费学习笔记（深入）”；
正确做法：每个需要并发的异步方法都独立使用 async with
每次并发需求都应开启新的 TaskGroup 上下文，而不是复用。这是最安全、最符合语义的方式。
- 每个需并发执行的
async def方法内，用async with asyncio.TaskGroup() as tg: - 不要试图提前创建、缓存或传递
TaskGroup实例 - 若多个方法需共享并发逻辑，提取为独立的 async 辅助函数，而非共享
TaskGroup
示例：
import asyncio
<p>class DataFetcher:
async def fetch_a(self) -> str:
await asyncio.sleep(0.1)
return "A"</p><pre class="brush:php;toolbar:false;"><code>async def fetch_b(self) -> str:
await asyncio.sleep(0.2)
return "B"
async def concurrent_fetch(self) -> tuple[str, str]:
async with asyncio.TaskGroup() as tg:
task_a = tg.create_task(self.fetch_a())
task_b = tg.create_task(self.fetch_b())
return task_a.result(), task_b.result()
立即学习“Python免费学习笔记（深入）”；
如何避免“TaskGroup is closed”错误？
这个错误几乎总是因为：在 async with 块外调用了 create_task，或者试图重复使用已退出的 TaskGroup 实例。
- 确保所有
tg.create_task(...)都严格写在async with ... as tg:缩进块内 - 不要把
tg保存为属性、传给其他协程、或在async with外 await 它 - 如果需要条件启动任务，把条件判断放在
async with块内（不是外面） - 注意：
TaskGroup不支持cancel()单个任务，只能整体取消整个组（通过外部asyncio.cancel()整个协程）
替代方案：什么时候该考虑 asyncio.gather 而不是 TaskGroup？
如果你不需要异常传播（即任一子任务失败就立刻终止全部）、也不需要在运行中动态增删任务，asyncio.gather 更轻量、写法更紧凑，且兼容性更好（3.7+）。
对比关键点：
-
TaskGroup：失败即停、支持中途添加、Python 3.11+、必须async with -
gather(return_exceptions=True)：默认全等、不中断、可捕获各任务异常、3.7+ - 性能差异极小，选型主要看错误处理语义是否匹配业务逻辑
例如只需并发跑几个固定任务且容错宽松：
async def simple_concurrent(self):
results = await asyncio.gather(
self.fetch_a(),
self.fetch_b(),
self.fetch_c()
)
return results
真正容易被忽略的是：TaskGroup 的异常传播是“精确中断”——它会让所有正在运行的子任务收到 CancelledError，但前提是这些子任务没有屏蔽取消信号。如果某个 fetch_x 内部用了 try/except CancelledError 或 asyncio.shield，TaskGroup 的失败传播就会失效，导致行为不符合预期。
