最近在找一个系统的bug，定位到了这个问题，虽然是AI修的。。但是还是很有启发，可以放在未来的rule里面来减少类似的bug。
0. 引言
在我们的日常编程中，资源管理是一个非常重要的环节。Python 提供了 try...finally 语句来确保资源在使用后能够被正确关闭。然而，finally 并不能处理所有的关闭情况，尤其是在涉及到异步编程和复杂的资源管理时。本文将深入探讨 finally 的使用场景、局限性，并介绍更优雅的资源管理方式。
1. finally 的基本使用
finally 块中的代码无论是否发生异常都会被执行，这使得它成为资源清理的理想选择。例如：
try:
resource = open('file.txt', 'r')
# 使用资源
finally:
resource.close()
在这个例子中，无论文件操作是否成功，resource.close() 都会被执行，从而确保文件被正确关闭。
2. finally 的局限性
尽管 finally 在大多数情况下都能很好地处理资源关闭，但它并不能覆盖所有的情况。例如，在异步编程中，finally 可能无法正确处理异步资源的关闭，尤其是在使用 asyncio 或其他异步框架时。以下是一些常见的局限性：
- 异步资源管理：在异步函数中，
finally代码块可能无法正确处理异步资源的关闭，因为它不会等待异步操作完成。 - 异常传播：如果在
finally块中抛出异常，它可能会覆盖原本的异常，导致原始异常信息丢失。 - 复杂的资源依赖：当多个资源相互依赖时，
finally块可能无法正确处理资源的关闭顺序，导致资源泄漏或未定义行为。
下面挨个来分析下
2.1 异步资源管理
我们分两种来看，一种是传统的 try...finally 模式，另一种是现代的 async with 模式。
2.1.1 finally方式关闭异步资源
定义两个函数，一个是用来模拟网络关闭的耗时操作，另一个是使用传统的 try...finally 模式来处理资源清理。
import asyncio
async def mock_network_close():
"""模拟一个耗时的清理操作（比如告诉数据库断开连接、发送离线状态等）"""
print(" -> [网络层] 正在向服务器发送断开协议...")
await asyncio.sleep(1) # 模拟网络耗时 1 秒
print(" -> [网络层] 断开协议发送成功，资源已安全释放！(如果看到这条，说明没泄露)")
async def bad_finally_worker():
"""使用传统的 try...finally 模式（脆弱的清理）"""
print("\n--- 1. 脆弱的 try...finally 模式 ---")
try:
print(" [Worker 1] 连接已建立，正在处理长时间任务...")
await asyncio.sleep(5)
except asyncio.CancelledError:
print(" [Worker 1] 收到强杀信号 (Cancel)！开始执行 finally 清理...")
raise
finally:
print(" [Worker 1] 进入 finally，准备关闭网络连接...")
# 坑点：如果在清理期间，外层（如框架、网关或超时机制）再次触发取消
# 这里的 await 就会被打断，抛出 CancelledError，导致清理逻辑中断！
await mock_network_close()
定义一个主函数来演示 finally 在异步环境下的局限性。我们将创建一个任务，并在任务运行一段时间后取消它，然后再次取消它以模拟“二次取消”的情况。
async def main():
# ==========================================
# 场景 1：演示 finally 的崩溃
# ==========================================
task1 = asyncio.create_task(bad_finally_worker())
await asyncio.sleep(1) # 让 worker 跑一会儿
# 第一次取消：比如用户断开了 websocket 连接
task1.cancel()
# 模拟极端但常见的情况：
# 系统见你取消后半天没断开，触发强制回收/或者上层的 asyncio.wait_for 发生了超时
# 这会导致针对同一个协程产生“二次取消”
await asyncio.sleep(0.1)
task1.cancel()
try:
await task1
except asyncio.CancelledError:
print(" [Main] Worker 1 彻底终止。注意上方：网络层资源泄漏了，没有看到'断开协议发送成功'！")
await asyncio.sleep(1) # 缓冲一下日志，方便观察
if __name__ == "__main__":
asyncio.run(main())
理论上日志输出如下：
--- 1. 脆弱的 try...finally 模式 ---
[Worker 1] 连接已建立，正在处理长时间任务...
[Worker 1] 收到强杀信号 (Cancel)！开始执行 finally 清理...
[Worker 1] 进入 finally，准备关闭网络连接...
-> [网络层] 正在向服务器发送断开协议...
[Main] Worker 1 彻底终止。注意上方：网络层资源泄漏了，没有看到'断开协议发送成功'！
可以看到，虽然 finally 块被执行了，但由于在清理过程中再次触发了取消操作，导致 await mock_network_close() 被中断，从而没有完成网络资源的关闭操作。这就说明了 finally 在异步环境下的局限性。
也就是你所依赖的finally没把你的屁股擦干净。。
2.1.2 使用 async with 处理异步资源
为了更优雅地处理异步资源的关闭，我们可以使用 async with 语句，它能够确保在退出上下文时正确地关闭资源，即使在异步环境中也能保证资源的释放。
同样，先定义一个上下文管理器来处理资源的安全关闭：
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def safe_connection():
"""使用现代的 async with 和上下文管理器（健壮的清理）"""
try:
yield
finally:
print("\n [ContextManager] 进入收尾阶段，准备关闭网络连接...")
# 现代异步框架底层的标准防抖做法：使用 asyncio.shield 保护关机协程
# shield 会吸收掉外部传来的 CancelledError，保证内部的 mock_network_close 依然能在后台跑完
try:
await asyncio.shield(mock_network_close())
except asyncio.CancelledError:
print(" [ContextManager] 即使护盾被打断，内部清理任务仍在后台坚强执行完！")
async def good_async_with_worker():
print("\n--- 2. 健壮的 async with 模式 ---")
async with safe_connection():
print(" [Worker 2] 连接已建立，正在处理长时间任务...")
await asyncio.sleep(5)
再定义一个主函数来演示 async with 的健壮性。我们将创建一个任务，并在任务运行一段时间后取消它，然后再次取消它以模拟“二次取消”的情况。
async def main():
# ==========================================
# 场景 2：演示 async with 的健壮性
# ==========================================
task2 = asyncio.create_task(good_async_with_worker())
await asyncio.sleep(1)
task2.cancel()
# 同样的二次取消打击
await asyncio.sleep(0.1)
task2.cancel()
try:
await task2
except asyncio.CancelledError:
print(" [Main] Worker 2 彻底终止。注意上方：即使被疯狂 Cancel，资源依旧被安全释放了！")
await asyncio.sleep(1.5) # 等待后台坚强跑完的护盾任务打印完毕
if __name__ == "__main__":
asyncio.run(main())
预期的日志输出如下：
--- 2. 健壮的 async with 模式 ---
[Worker 2] 连接已建立，正在处理长时间任务...
[ContextManager] 进入收尾阶段，准备关闭网络连接...
-> [网络层] 正在向服务器发送断开协议...
[ContextManager] 即使护盾被打断，内部清理任务仍在后台坚强执行完！
[Main] Worker 2 彻底终止。注意上方：即使被疯狂 Cancel，资源依旧被安全释放了！
-> [网络层] 断开协议发送成功，资源已安全释放！(如果看到这条，说明没泄露)
2.2 异常传播问题
这是 Python 最臭名昭著的坑之一（同步异步都存在）
try:
1 / 0 # 发生 ZeroDivisionError (真正的原始错误)
finally:
# 试图清理资源，但网络突然断了
await session.close() # 发生 TimeoutError
在这个例子中，业务逻辑是因为“除以零”崩溃的，这才是你需要排查的 Bug。
但是，当程序进入 finally 准备善后时，因为网络问题抛出了一个 TimeoutError。
在 Python 引擎的规则里：finally 里面新产生的异常，会直接“吞掉”并覆盖掉 try 里面的原始异常。
结果就是：你在控制台或者日志系统里，只会看到 TimeoutError，而那个致命的 ZeroDivisionError 就这么凭空消失了！这会让你在查 Bug 时完全摸不着头脑。
解决方案： 优秀的 async with 上下文管理器，在底层的 aexit 方法中，会精细地处理多重异常（通过 context 和 cause 将多个异常串联起来），确保原始的业务异常永远不会被清理操作的异常给掩盖。
2.3 复杂的资源依赖问题
假设你在一个函数里打开了三个资源：1. 数据库连接、2. Redis 连接、3. 本地日志文件。 并且它们有依赖关系：关闭时必须先关本地文件，再关 Redis，最后关数据库。
如果你用 finally 来写，代码会变成可怕的“俄罗斯套娃”（Arrow Anti-Pattern）：
try:
db = await connect_db()
try:
redis = await connect_redis()
try:
file = open("log.txt")
# --- 业务逻辑 ---
finally:
file.close() # 必须最先关
finally:
await redis.close()
finally:
await db.close() # 必须最后关
如果不这么嵌套，只要中间某一个 close() 报错了（见上面的第二点），后面的 close() 就永远不会执行，直接导致资源泄漏。
解决方案： 使用 async with，你可以极其优雅、平铺地解决依赖顺序问题。Python 会自动保证它们以栈（Stack）的形式先进后出（LIFO），即最后打开的资源最先被安全关闭，即使中间报错，也会稳妥地把剩下的资源关掉：
async with connect_db() as db, \
connect_redis() as redis, \
open_file("log.txt") as file:
# --- 业务逻辑 ---
# 离开时，系统自动安全关闭：file -> redis -> db
3. 总结
finally在同步编程中是一个非常有用的工具，但在异步编程中，它的局限性显而易见，尤其是在处理异步资源和复杂的资源依赖时。- 使用
async with和上下文管理器可以更优雅地处理资源关闭，确保即使在异常或取消的情况下，资源也能被正确释放。 - 在设计异步程序时，应该尽量避免依赖
finally来管理资源，而是使用现代的异步上下文管理器来确保资源的安全关闭和异常的正确传播。