使用TaskGroup时，如何在任务完成后立即处理结果？

TaskGroup的核心优势是安全管理任务生命周期（自动取消未完成任务、统一处理异常），但默认确实会等待所有任务结束。要实现类似asyncio.as_completed()或FIRST_COMPLETED的效果，完全可以把TaskGroup和队列/回调结合起来，分享两个实用方案：

方法1：用asyncio.Queue实时接收任务结果 #

在TaskGroup中启动任务，每个任务完成后把结果塞进队列，主线程监听队列，拿到第一个结果就开始处理，剩下的任务仍由TaskGroup管理（可按需手动取消）：

import asyncio

async def task_func(id, delay):
    await asyncio.sleep(delay)
    return f"任务{id}完成"

async def main():
    result_queue = asyncio.Queue()
    
    async with asyncio.TaskGroup() as tg:
        # 启动多个任务，完成后将结果放入队列
        for i in range(3):
            tg.create_task(
                lambda i=i: result_queue.put_nowait(await task_func(i, i+1))
            )
        
        # 监听队列，获取首个结果并处理
        first_result = await result_queue.get()
        print(f"首个完成的任务结果：{first_result}")
        
        # 若不需要剩余任务，可取消整个TaskGroup
        tg.cancel()

asyncio.run(main())

方法2：给任务绑定回调函数 #

利用Task.add_done_callback()，让每个任务完成后立即触发处理逻辑，首个完成的任务会第一时间执行回调，同时TaskGroup依然负责后续任务的生命周期管理：

import asyncio

async def task_func(id, delay):
    await asyncio.sleep(delay)
    return f"任务{id}完成"

def handle_result(task):
    result = task.result()
    print(f"收到任务结果：{result}")
    # 标记首个完成的任务并执行特定逻辑
    global first_handled
    if not first_handled:
        print("这是首个完成的任务，启动后续处理流程")
        first_handled = True

async def main():
    global first_handled
    first_handled = False
    
    async with asyncio.TaskGroup() as tg:
        for i in range(3):
            task = tg.create_task(task_func(i, i+1))
            task.add_done_callback(handle_result)

asyncio.run(main())

没错，它不是严格意义上的全面替代，而是一种设计权衡。TaskGroup的目标是解决create_task()手动管理时易出现的漏洞（比如遗漏任务取消、异常未传播导致的“幽灵任务”），核心定位是“批量管理任务的完整生命周期”，因此默认会等待所有任务结束。

如果你的需求是“提前处理结果”这类场景，它确实不如as_completed()或wait()直接，但这不代表它没有价值——你可以将它和这些工具结合使用，既享受TaskGroup的安全性，又实现所需的异步流程。

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