如何让asyncio任务触发TimeoutError前自行处理超时异常

在Python 3.9中，asyncio.wait_for超时后会自动取消目标异步任务，你可以在eternity函数内部通过捕获asyncio.CancelledError或者使用finally块来执行资源清理逻辑——因为只有任务自身知晓需要清理的资源，这种方式能确保清理操作在任务被中断时触发。

核心实现思路 #

1. asyncio.wait_for超时后会调用任务的cancel()方法，触发任务内部的CancelledError
2. 在eternity的try/except块中捕获该异常，执行专属清理逻辑
3. 可选：使用finally块保证无论任务正常结束还是被取消，都执行必要的资源释放

代码示例 #

import asyncio

async def eternity():
    # 模拟业务中创建的资源（比如打开的文件、网络连接、临时文件等）
    business_resource = "用户会话连接/临时缓存文件"
    print(f"初始化资源: {business_resource}")

    try:
        # 模拟多步长时间异步操作
        for step in range(1, 6):
            print(f"执行业务步骤 {step}")
            await asyncio.sleep(2)  # 每步耗时2秒，总时长超过5秒超时阈值
    except asyncio.CancelledError:
        # 捕获取消信号，执行资源清理
        print("任务超时被取消，开始清理资源")
        # 这里替换为实际的清理逻辑：关闭连接、删除临时文件、释放锁等
        print(f"已清理资源: {business_resource}")
        # 重新抛出异常，让外层wait_for触发TimeoutError（可选，根据业务需求决定）
        raise
    finally:
        # 若有无论如何都要执行的清理，可放在此处（比如兜底清理）
        # print("执行兜底资源清理")
        pass

async def main():
    try:
        await asyncio.wait_for(eternity(), timeout=5)
        print("任务正常执行完成")
    except asyncio.TimeoutError:
        print("任务超时（主线程提示）")

asyncio.run(main())

关键说明 #

• 捕获CancelledError后，你可以根据业务需求决定是否重新抛出异常：如果需要让主线程感知到超时，重新抛出即可，wait_for会将其转换为TimeoutError；如果不需要主线程处理，也可以不抛出。
• 若资源必须确保被清理（比如文件句柄），优先使用finally块，它会在协程结束（无论正常还是异常）时执行。
• 不要在清理逻辑中执行耗时操作，避免阻塞事件循环。

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