Python AsyncExitStack 怎么管理异步资源：多连接清理、异常传播与退出顺序

异步任务里最容易漏掉的，不是打开资源，而是中途失败后到底谁负责关闭。连接数量由配置决定、某个资源只在特定分支才创建时，连续写几个 try/finally 很快就会变成一张难以维护的嵌套网。Python 的 contextlib.AsyncExitStack 可以把这些异步上下文管理器登记到同一个栈里，退出时按后进先出顺序清理；资源越晚加入，越早释放。

先看基线：动态资源为什么容易漏清理

假设一个批处理请求按租户配置打开若干异步连接。连接数为 0、打开第 2 个连接失败、业务体抛出异常，都是合法路径。手写嵌套结构时，常见问题是只给“全部打开成功”写了关闭逻辑，或者某个新分支忘记补一层 finally。

这里先别急着把所有资源都改成全局池。真正要理清的是资源所有权：谁成功拿到资源，谁就必须把清理动作登记到一个仍然可访问的退出栈里。

最小写法：enter_async_context 统一登记资源

异步上下文管理器可以直接交给 enter_async_context()。它会先执行资源的异步进入方法，成功后把对应的退出方法压入栈；如果进入阶段抛出异常，尚未成功进入的资源不会被误关闭。

from contextlib import AsyncExitStack

class AsyncConnection:
    def __init__(self, name, events):
        self.name = name
        self.events = events

    async def __aenter__(self):
        self.events.append(f"open:{self.name}")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self.events.append(f"close:{self.name}")
        return False

async def load_connections(names, events):
    async with AsyncExitStack() as stack:
        connections = []
        for name in names:
            conn = await stack.enter_async_context(
                AsyncConnection(name, events)
            )
            connections.append(conn)
        return [conn.name for conn in connections]

调用 load_connections(["primary", "audit"], events) 后，events 应该先记录两个打开动作，再按 close:audit、close:primary 的顺序收尾。这个顺序不是装饰性的：审计连接若依赖主连接仍然存在，就应该先释放审计连接。

没有上下文协议时，用 push_async_callback 补上所有权

有些客户端只返回一个带 aclose() 方法的对象，并没有实现 __aenter__ 与 __aexit__。这时可以把异步清理函数登记到栈里。关键是登记动作要紧跟在“确认资源已经拿到”之后，不能等业务处理完才补。

async def load_client(factory, events):
    async with AsyncExitStack() as stack:
        client = await factory()
        stack.push_async_callback(client.aclose)
        events.append("client:ready")
        return await client.fetch()

如果 factory() 失败，aclose 不会被登记；如果 fetch() 失败，退出栈仍会调用它。异步回调最好保持无参数，若清理需要固定参数，可以用一个闭包明确捕获资源实例，避免把后续可变状态带进去。

用可复现指标验收退出顺序和清理覆盖率

性能和资源管理不能只看最终返回值。可以在测试夹具里记录 open、close 事件，并统计每条路径的打开数、关闭数和未配对数。下面的断言覆盖成功、业务异常和中途打开失败三个边界：

import pytest

@pytest.mark.asyncio
async def test_stack_closes_in_reverse_order():
    events = []
    names = await load_connections(["primary", "audit"], events)

    assert names == ["primary", "audit"]
    assert events == [
        "open:primary", "open:audit",
        "close:audit", "close:primary",
    ]

在本地压测中，可以把资源工厂分别设置为 0、1、8 和 32 个，记录每轮 opened - closed 的差值。理想结果始终为 0；如果差值只在异常路径出现，优先检查资源是否在成功创建后立即登记，而不是先扩大连接池。

边界条件：清理异常、提前 pop 和 cancel

清理回调抛异常怎么办？

退出栈会继续处理已经登记的其他退出回调，但最终退出结果会受到清理异常影响。生产代码应记录资源名和原始异常，不要吞掉业务本身抛出的异常；对必须确保尽力清理的连接，可以在清理函数内部把可预期的网络断开异常转换为常规日志事件。

什么时候应该 pop_all？

只有当资源所有权要转移给另一个明确的生命周期管理者时，才考虑 pop_all()。普通的“暂时不想清理”不是理由，否则调用方会以为上下文退出已经完成释放。

任务被取消时还能清理吗？

async with AsyncExitStack() 的退出路径仍会执行，但清理回调本身也可能被取消。对必须完成的极短收尾，可以在清理函数中谨慎使用屏蔽取消的策略，并设置自己的时间上限，不能无限等待。

把规则收成一张上线前检查表

• 每个资源成功获得后立即进入 enter_async_context() 或 push_async_callback()。
• 测试资源数为 0、部分打开失败、业务异常和任务取消这几类场景。
• 断言所有路径的 opened == closed，并核对逆序释放。
• 清理异常要保留资源名和原始异常，不要用宽泛的 except Exception 静默吞掉。

常见问题

AsyncExitStack 只能管理异步资源吗？

不是。同步资源可以使用 ExitStack；一个栈内混用时要明确选择对应的进入方法，避免把同步对象误交给异步协议。

资源打开失败后会不会关闭一个没有打开成功的对象？

不会。只有成功进入上下文并完成登记的资源才会出现在退出链中，所有已成功打开的资源都会按登记的逆序完成清理。

为什么不用多个 finally 代替 AsyncExitStack？

资源数量固定、生命周期简单时多个 finally 完全可以；当资源由配置或运行时分支决定时，AsyncExitStack 能把所有权登记和退出顺序集中起来，更容易覆盖异常路径。

AsyncExitStack 的价值不是少写几行缩进，而是把「谁获得资源、谁负责释放、异常时按什么顺序释放」变成可观测的运行规则。把打开数、关闭数和事件顺序写进测试用例，动态资源场景才能真正做到可控。