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:
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 继续盲跑或卡死。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成,提供 184 种工具和 200 多条命令,用于安全分析和代码质量...
立即学习“Python免费学习笔记(深入)”;
- 若 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() 的位置就会出错。

















