asyncio异常处理问题：create_task任务异常无法被gather捕获

asyncio.create_task() 创建的是独立的顶级任务，不属于 asyncio.gather() 跟踪的任务集合。因此，该子任务的异常不会冒泡到 gather() 所在的 try/except 上下文，默认情况下未处理的任务异常会触发事件循环的全局异常处理逻辑（Python 3.7+ 会直接终止事件循环）。

以下是几种可行的处理方式：

1. 给子任务添加完成回调捕获异常 #

通过 add_done_callback 为子任务绑定异常处理函数，在任务完成时主动检查并处理异常：

import asyncio

async def coro3():
    raise RuntimeError("coro3抛出的异常")

async def coro1():
    def handle_coro3_exception(task):
        try:
            task.result()  # 获取任务结果，触发异常
        except Exception as e:
            print(f"捕获到coro3的异常: {e}")
            # 这里可添加后续处理逻辑，例如终止其他协程、记录日志等

    # 创建子任务并绑定回调
    task = asyncio.create_task(coro3())
    task.add_done_callback(handle_coro3_exception)
    
    while True:
        print("coro1")
        await asyncio.sleep(1)
        
async def coro2():
    while True:
        print("coro2")
        await asyncio.sleep(1)

async def main():
    try:
        await asyncio.gather(coro1(), coro2())
    except Exception as e:
        print(f"捕获到协程异常: {e}")

asyncio.run(main())

2. 使用 asyncio.TaskGroup（Python 3.11+ 推荐） #

TaskGroup 是 Python 3.11 引入的更安全的任务管理工具，它会自动跟踪所有在其上下文内创建的任务（包括嵌套任务），任何任务的异常都会传播到 TaskGroup 的上下文中，可直接被 try/except 捕获：

import asyncio

async def coro3():
    raise RuntimeError("coro3抛出的异常")

async def coro1():
    # 在TaskGroup上下文内创建的子任务会被自动跟踪
    async with asyncio.TaskGroup() as tg:
        tg.create_task(coro3())
        while True:
            print("coro1")
            await asyncio.sleep(1)
        
async def coro2():
    while True:
        print("coro2")
        await asyncio.sleep(1)

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(coro1())
            tg.create_task(coro2())
    except Exception as e:
        print(f"捕获到异常: {e}")

asyncio.run(main())

当 coro3 抛出异常时，TaskGroup 会立即终止所有关联任务，并将异常传播到 main 的 try/except 块中。

3. 全局事件循环异常处理（不推荐） #

通过设置事件循环的全局异常处理函数，可捕获所有未处理的任务异常，但此方法是全局生效，无法针对特定上下文做差异化处理：

import asyncio

def global_exception_handler(loop, context):
    print(f"全局捕获异常: {context['exception']}")

async def coro3():
    raise RuntimeError("coro3抛出的异常")

async def coro1():
    asyncio.create_task(coro3())
    while True:
        print("coro1")
        await asyncio.sleep(1)
        
async def coro2():
    while True:
        print("coro2")
        await asyncio.sleep(1)

async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(global_exception_handler)
    await asyncio.gather(coro1(), coro2())

asyncio.run(main())

如果你的 Python 版本是 3.11+，优先使用 TaskGroup，它提供了更简洁安全的任务管理和异常传播机制；若需兼容旧版本，使用子任务回调的方式更灵活，能针对性处理特定任务的异常。

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