如何让asyncio.gather在任务出错时立即抛出异常？

你的代码里使用了time.sleep(1)这种同步阻塞操作，它会占用全局解释器锁（GIL）并阻塞整个事件循环。当第一个任务抛出RuntimeError后，事件循环因为被阻塞无法及时切换到asyncio.gather去处理异常，必须等所有任务的阻塞操作全部执行完毕，才能处理任务的结果和异常，这就是为什么要等10秒才抛出异常的原因。

要实现"单个任务异常时立即传播并终止其他任务"，需要两步：

1. 将CPU密集/阻塞操作异步化，避免卡住事件循环；
2. 监控任务执行状态，第一个异常出现时立即取消其他任务并抛出异常。

完整代码示例 #

import asyncio
import time


async def main():
    num_reps = 10
    counter = 0

    async def generate():
        nonlocal counter
        # 用asyncio.to_thread将阻塞操作放到线程池执行（Python3.9+支持）
        # 避免阻塞事件循环，让事件循环能及时调度任务状态
        await asyncio.to_thread(time.sleep, 1)
        counter += 1
        print(counter, end="\r")
        raise RuntimeError  # 模拟计算出错

    # 创建所有任务（必须用create_task，否则是协程对象不会立即执行）
    tasks = [asyncio.create_task(generate()) for _ in range(num_reps)]
    
    # 等待第一个完成的任务（包括异常完成）
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
    
    # 检查完成的任务是否有异常
    for task in done:
        try:
            # 获取任务结果，如果有异常会重新抛出
            task.result()
        except Exception as e:
            # 取消所有未完成的任务
            for t in pending:
                t.cancel()
            # 传播异常
            raise e
    
    # 如果没有异常，收集所有任务结果（可选）
    results = await asyncio.gather(*tasks)
    return results


asyncio.get_event_loop().run_until_complete(main())

关键说明 #

• asyncio.to_thread：把同步阻塞的time.sleep放到线程池执行，让事件循环能正常调度其他任务和处理状态变更，这是解决问题的核心前提。
• asyncio.wait + FIRST_EXCEPTION：告诉事件循环只要有一个任务异常完成，就立即返回，不用等所有任务结束。
• 取消未完成任务：当捕获到第一个异常时，主动取消所有还在运行的任务，避免无效计算继续占用资源。

这样修改后，第一个任务抛出异常时，事件循环能立即感知到，取消其他任务并马上传播异常，整个过程只会等待1秒左右，而不是10秒。

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