最近在找一个系统的bug，定位到了这个问题，虽然是AI修的。。但是还是很有启发，可以放在未来的rule里面来减少类似的bug。
在我们的日常编程中，资源管理是一个非常重要的环节。Python 提供了 try...finally 语句来确保资源在使用后能够被正确关闭。然而，finally 并不能处理所有的关闭情况，尤其是在涉及到异步编程和复杂的资源管理时。本文将深入探讨 finally 的使用场景、局限性，并介绍更优雅的资源管理方式。
finally 块中的代码无论是否发生异常都会被执行，这使得它成为资源清理的理想选择。例如：
try:
resource = open('file.txt', 'r')
finally:
resource.close()
在这个例子中，无论文件操作是否成功，resource.close() 都会被执行，从而确保文件被正确关闭。
尽管 finally 在大多数情况下都能很好地处理 资源关闭，但它并不能覆盖所有的情况。例如，在异步编程中，finally 可能无法正确处理异步资源的关闭，尤其是在使用 asyncio 或其他异步框架时。以下是一些常见的局限性：
- 异步资源管理：在异步函数中，
finally 代码块可能无法正确处理异步资源的关闭，因为它不会等待异步操作完成。
- 异常传播：如果在
finally 块中抛出异常，它可能会覆盖原本的异常，导致原始异常信息丢失。
- 复杂的资源依赖：当多个资源相互依赖时，
finally 块可能无法正确处理资源的关闭顺序，导致资源泄漏或未定义行为。
下面挨个来分析下
我们分两种来看，一种是传统的 try...finally 模式，另一种是现代的 async with 模式。
定义两个函数，一个是用来模拟网络关闭的耗时操作，另一个是使用传统的 try...finally 模式来处理资源清理。
import asyncio
async def mock_network_close():
"""模拟一个耗时的清理操作（比如告诉数据库断开连接、发送离线状态等）"""
print(" -> [网络层] 正在向服务器发送断开协议...")
await asyncio.sleep(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 mock_network_close()
定义一个主函数来演示 finally 在异步环境下的局限性。我们将创建一个任务，并在任务运行一段时间后取消它，然后再次取消它以模拟“二次取消”的情况。
async def main():
task1 = asyncio.create_task(bad_finally_worker())
await asyncio.sleep(1)
task1.cancel()
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没把你的屁股擦干净。。
为了更优雅地处理异步资源的关闭，我们可以使用 async with 语句，它能够确保在退出上下文时正确地关闭资源，即使在异步环境中也能保证资源的释放。
同样，先定义一个上下文管理器来处理资源的安全关闭：
import asyncio
from contextlib import asynccontextmanager
@asynccontextmanager
async def safe_connection():
"""使用现代的 async with 和上下文管理器（健壮的清理）"""
try:
yield
finally:
print("\n [ContextManager] 进入收尾阶段，准备关闭网络连接...")
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():
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，资源依旧被安全释放了！
-> [网络层] 断开协议发送成功，资源已安全释放！(如果看到这条，说明没泄露)
这是 Python 最臭名昭著的坑之一（同步异步都存在）
try:
1 / 0
finally:
await session.close()
在这个例子中，业务逻辑是因为“除以零”崩溃的，这才是你需要排查的 Bug。
但是，当程序进入 finally 准备善后时，因为网络问题抛出了一个 TimeoutError。
在 Python 引擎的规则里：finally 里面新产生的异常，会直接 “吞掉”并覆盖掉 try 里面的原始异常。
结果就是：你在控制台或者日志系统里，只会看到 TimeoutError，而那个致命的 ZeroDivisionError 就这么凭空消失了！这会让你在查 Bug 时完全摸不着头脑。
解决方案： 优秀的 async with 上下文管理器，在底层的 aexit 方法中，会精细地处理多重异常（通过 context 和 cause 将多个异常串联起来），确保原始的业务异常永远不会被清理操作的异常给掩盖。
假设你在一个函数里打开了三个资源：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:
finally 在同步编程中是一个非常有用的工具，但在异步编程中，它的局限性显而易见，尤其是在处理异步资源和复杂的资源依赖时。- 使用
async with 和上下文管理器可以更优雅地处理资源关闭，确保即使在异常或取消的情况下，资源也能被正确释放。
- 在设计异步程序时，应该尽量避免依赖
finally 来管理资源，而是使用现代的异步上下文管理器来确保资源的安全关闭和异常的正确传播。
