taskgroup不能直接await queue.get()，因其返回协程需在事件循环中显式驱动；正确做法是封装worker协程，在async with taskgroup中启动，并确保每个queue.get()后调用queue.task_done()。
TaskGroup 无法直接 await queue.get()
因为 asyncio.Queue.get() 返回的是协程对象，而 asyncio.TaskGroup 的 create_task() 接收的是协程对象 —— 听起来能用，但实际会出错：如果你直接把 queue.get() 传给 create_task()，它会在 Task 启动前就尝试 await（隐式调用），导致 RuntimeError：RuntimeError: await was called on a coroutine object created by "get()", but the event loop is not running。根本原因是 queue.get() 协程必须在事件循环中被驱动，而 TaskGroup.create_task() 不会自动 await 它；你得显式 await 或包装成可调度的协程。
正确写法：用 async for 或 while True + await get()
消费队列任务的主体逻辑必须是 async 函数，且内部主动 await queue.get()。常见安全模式是封装成 worker 协程：
async def worker(queue: asyncio.Queue):
while True:
try:
item = await queue.get()
# 处理 item
await asyncio.sleep(0.1) # 模拟异步工作
except asyncio.CancelledError:
break
finally:
queue.task_done()
然后在 async with asyncio.TaskGroup() as tg: 中启动多个 worker：
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成，提供 184 种工具和 200 多条命令，用于安全分析和代码质量...
async def main():
queue = asyncio.Queue()
async with asyncio.TaskGroup() as tg:
for _ in range(3):
tg.create_task(worker(queue))
# 生产者逻辑（可选）
for i in range(5):
await queue.put(i)
await queue.join() # 等待所有已入队任务完成
-
queue.task_done()必须在每个成功处理后调用，否则queue.join()永远不会返回 - worker 中用
try/except CancelledError是为了响应 TaskGroup 取消信号，避免残留任务 - 不要在 worker 外部用
await queue.get()后再 create_task —— 那样只消费一个，不是“持续消费”
为什么不用 asyncio.create_task() 而用 TaskGroup？
关键区别在于错误传播和生命周期管理：asyncio.create_task() 启动的任务出错会被静默吞掉（除非你手动检查 task.exception()），而 TaskGroup 遇到任一子任务抛出未捕获异常，会立即取消其余任务并重新抛出该异常 —— 这对队列消费场景至关重要。比如某个 worker 因网络错误崩溃，你不希望其他 worker 继续盲跑或卡死。
- 若 worker 内部有未处理的
Exception（非CancelledError），TaskGroup会立刻终止全部 worker 并 raise 原始异常 - Python 3.11+ 的
TaskGroup是结构化并发的标准方式，比手写asyncio.gather(..., return_exceptions=False)更清晰、更安全 - 注意：
TaskGroup本身不提供“任务重试”或“死信队列”，这些需自行在 worker 内实现
queue.join() 和 task_done() 的配合容易漏掉
这是最常踩的坑：很多人记得 await queue.join()，却忘了在每个 queue.get() 对应的处理完成后调用 queue.task_done()。一旦漏掉，join() 就永远阻塞，整个 async with TaskGroup() 也无法退出。
-
queue.task_done()必须与queue.get()一一对应，哪怕处理失败也要在finally块里调用（除非你想让失败任务一直卡住队列） - 如果 worker 中用了
break退出循环，要确保退出前已处理完当前 item 并调用task_done() - 调试时可临时加
print(f"done {item}, qsize={queue.qsize()}")观察是否匹配
实际项目里，真正麻烦的往往不是启动几个 worker，而是如何定义“任务完成”的语义：是成功写入数据库？还是 HTTP 请求返回 2xx？这个边界一旦模糊，task_done() 的位置就会出错。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
