如何在asyncio.gather中正确处理SystemExit异常，使程序返回0退出码

你遇到的问题核心在于SystemExit是Python的特殊基类异常，asyncio对它的处理逻辑和普通异常不同——即使你用return_exceptions=True捕获了它，当asyncio.run()在退出阶段调用_cancel_all_tasks()时，这个异常依然会被传播，最终导致进程退出码为1。这是因为_cancel_all_tasks()会遍历所有关联到事件循环的任务（包括已完成的），并通过asyncio.gather()等待它们，而SystemExit即使被return_exceptions=True包裹，依然会触发asyncio的顶层异常处理逻辑。

下面是几种不需要直接在func内部包装SystemExit的解决方案：

方案1：自定义替代asyncio.run()的函数，仅处理未完成任务 #

asyncio.run()默认的_cancel_all_tasks()会处理所有任务，包括已完成的。我们可以自己实现一个简化版的运行函数，只取消并等待未完成的任务，避免触发已完成任务中的SystemExit异常：

import asyncio

async def func(t):
    await asyncio.sleep(t)
    raise SystemExit('Err')

def my_asyncio_run(main, *, debug=False):
    loop = asyncio.new_event_loop()
    try:
        asyncio.set_event_loop(loop)
        if debug:
            loop.set_debug(True)
        return loop.run_until_complete(main())
    finally:
        # 仅处理未完成的任务，跳过已完成的任务
        pending_tasks = [task for task in asyncio.all_tasks(loop) if not task.done()]
        if pending_tasks:
            for task in pending_tasks:
                task.cancel()
            # 用return_exceptions=True确保取消过程中不会抛出异常
            loop.run_until_complete(asyncio.gather(*pending_tasks, return_exceptions=True))
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.close()

async def main():
    tasks = [asyncio.create_task(func(t)) for t in range(0, 3)]
    print('Starting gather')
    await asyncio.gather(*tasks, return_exceptions=True)
    print('Gather returned')
    # 清理未完成的任务（可选，因为my_asyncio_run会处理）
    for t in tasks:
        if not t.done():
            t.cancel()
            try:
                await t
            except BaseException:
                pass
    await asyncio.sleep(1)

my_asyncio_run(main())

方案2：通过任务回调提前“消耗”异常 #

给每个任务添加一个done_callback，在任务完成时主动调用task.exception()获取异常，这样asyncio就不会把它标记为“未处理异常”，后续即使被_cancel_all_tasks()等待，也不会触发进程退出：

import asyncio

async def func(t):
    await asyncio.sleep(t)
    raise SystemExit('Err')

def consume_exception(task):
    # 主动获取异常，避免asyncio将其视为未处理异常
    task.exception()

async def main():
    tasks = []
    for t in range(0, 3):
        task = asyncio.create_task(func(t))
        task.add_done_callback(consume_exception)
        tasks.append(task)
    
    print('Starting gather')
    await asyncio.gather(*tasks, return_exceptions=True)
    print('Gather returned')
    
    for t in tasks:
        if not t.done():
            t.cancel()
            try:
                await t
            except BaseException:
                pass
    await asyncio.sleep(1)

asyncio.run(main())

方案3：用装饰器间接转换SystemExit为普通异常 #

虽然你不想直接修改func内部，但可以通过装饰器统一处理，不需要侵入原函数逻辑：

import asyncio
from functools import wraps

def wrap_system_exit(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except SystemExit as e:
            # 将SystemExit转换为普通RuntimeError，避免asyncio的特殊处理
            raise RuntimeError(str(e)) from e
    return wrapper

@wrap_system_exit
async def func(t):
    await asyncio.sleep(t)
    raise SystemExit('Err')

async def main():
    tasks = [asyncio.create_task(func(t)) for t in range(0, 3)]
    print('Starting gather')
    await asyncio.gather(*tasks, return_exceptions=True)
    print('Gather returned')
    for t in tasks:
        if not t.done():
            t.cancel()
            try:
                await t
            except BaseException:
                pass
    await asyncio.sleep(1)

asyncio.run(main())

方案有效性说明 #

• 方案1通过跳过已完成任务，彻底避免了_cancel_all_tasks()触发SystemExit的传播；
• 方案2通过主动获取异常，让asyncio认为异常已经被处理，不会在顶层触发退出逻辑；
• 方案3将特殊的SystemExit转换为普通异常，让return_exceptions=True可以正常捕获，不会触发asyncio的特殊处理。

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