在现代 Python 异步编程中，async with 语法被广泛用于管理资源的生命周期，例如文件、网络连接、数据库会话等。当我们需要动态管理多个异步资源，或者避免层层嵌套的 async with 时，contextlib.AsyncExitStack 就会成为一个非常棒的助手。

---

一、什么是 AsyncExitStack？

AsyncExitStack 是 Python 标准库 contextlib 提供的一个类，用于动态地注册并管理多个异步上下文管理器。它的行为类似于 async with 的堆栈版，但比固定嵌套结构更灵活，适用于资源数目或类型不确定的情况。

简而言之，它是一个可以在运行时动态注册多个 async with 对象，并在退出时自动逆序释放所有资源的工具。

二、为什么需要 AsyncExitStack？

场景一：不知道需要多少个异步资源

比如要并发连接一组服务，服务列表在运行时才知道。

for config in server_configs:
    async with connect_to_server(config) as conn:
        ...

不能写成固定嵌套，因为我们根本不知道有几个。

场景二：希望提前中断执行，但又能保证资源全部释放

如果中途抛异常、提前 return，普通 async with 很难做到“统一退出逻辑”。

场景三：希望将多个 async with 封装在循环、函数中，而不是嵌套结构

三、基本用法

from contextlib import AsyncExitStack

async with AsyncExitStack() as stack:
    conn1 = await stack.enter_async_context(connect_to_server("A"))
    conn2 = await stack.enter_async_context(connect_to_server("B"))

    print("执行任务中...")

# 这里 conn1、conn2 都会被自动关闭，顺序为 B -> A

退出 async with 块时，所有注册的上下文管理器会以相反顺序（先进后出）执行 __aexit__，确保资源按注册顺序清理。

【注意】Python 要求所有异步语法如 await、async with、async for 都必须在 async def 函数体中使用。

---

四、示例

1、自定义异步资源类示例

class MyAsyncContext:
    async def __aenter__(self):
        print("连接资源")
        return self

    async def __aexit__(self, exc_type, exc_val, tb):
        print("释放资源")

使用方式：

from contextlib import AsyncExitStack
import asyncio

async def main():
    async with AsyncExitStack() as stack:
        res1 = await stack.enter_async_context(MyAsyncContext())
        res2 = await stack.enter_async_context(MyAsyncContext())
        print("处理中...")


if __name__ == "__main__":
    asyncio.run(main())

# 运行结果：
# 连接资源
# 连接资源
# 处理中...
# 释放资源
# 释放资源

2、动态连接多个异步服务

from contextlib import AsyncExitStack
import asyncio


async def connect(name):
    print(f"[{name}] 已连接")
    return name


async def disconnect(name):
    print(f"[{name}] 已断开")


async def connect_wrapper(name):
    class Conn:
        async def __aenter__(self):
            await connect(name)
            return name

        async def __aexit__(self, exc_type, exc, tb):
            await disconnect(name)

    return Conn()


async def main():
    services = ["redis", "mongo", "mysql"]

    async with AsyncExitStack() as stack:
        connections = []
        for name in services:
            conn = await stack.enter_async_context(await connect_wrapper(name))
            connections.append(conn)

        print("处理业务中...")


if __name__ == "__main__":
    asyncio.run(main())

# 运行结果：
# [redis] 已连接
# [mongo] 已连接
# [mysql] 已连接
# 处理业务中...
# [mysql] 已断开
# [mongo] 已断开
# [redis] 已断开

✅ 所有连接在退出时会自动按注册顺序反向断开，哪怕中途抛出异常也能保证资源释放！

五、对比传统 async with 嵌套写法

❌ 普通嵌套（写死结构）

async with A() as a:
    async with B() as b:
        async with C() as c:
            ...

• 不可动态配置
• 可读性差
• 缺乏统一清理机制

✅ AsyncExitStack（动态 + 清晰 + 安全）

async with AsyncExitStack() as stack:
    a = await stack.enter_async_context(A())
    b = await stack.enter_async_context(B())
    c = await stack.enter_async_context(C())

✅ 使用建议与注意事项

建议	原因
尽量配合异常处理使用	保证在出错时资源也能清理
不同资源使用不同 stack	便于精细化管理
清理不支持 async 的对象时可用 push_async_callback()（详情可见文章底部的补充部分）	比如手动关闭连接、释放句柄等

✅ 总结

特性	描述
灵活性	动态管理任意数量的异步资源
安全性	异常情况下也能确保资源被正确释放
可读性	避免深层嵌套，逻辑更清晰
常见场景	多连接管理、任务链式清理、代理生命周期控制等

✅ 适用场景

应用场景	是否推荐使用 AsyncExitStack
需要动态注册多个异步资源	✅✅✅
资源清理需要统一处理	✅✅✅
只管理一个静态上下文	❌
希望支持失败回滚清理逻辑	✅✅✅

当我们在写多异步连接、任务链清理、或实现类似“浏览器多标签管理、WebSocket 会话管理”这类组件时，AsyncExitStack 就成为了不可多得的帮手。它比 async with 更强大、更灵活、更安全。

如果大家熟悉事务管理、数据库连接池、堆栈资源释放，那么就能更快理解它的设计哲学：
把所有“需要清理”的事统一堆栈管理，最后一次性释放，确保万无一失。

---

补充：push_async_callback()方法

当我们需要在 async with 块中清理某些“不支持 async 上下文协议”的资源时（比如普通的关闭函数），我们可以用 push_async_callback() 注册一个异步的清理函数。

1、为什么需要它？

有时我们要管理的资源并没有 __aenter__ / __aexit__ 方法，也就是说它不能用 async with 来管理，但我们又希望退出时能自动异步清理这些资源，这时就需要用：

stack.push_async_callback(你的清理函数, 可选参数...)

2、举个例子：连接 WebSocket 后用 async close 清理

from contextlib import AsyncExitStack
import asyncio

class DummyWS:
    async def close(self):
        print("🔌 WebSocket 关闭连接")

async def main():
    ws = DummyWS()

    async with AsyncExitStack() as stack:
        # 注册异步清理函数
        stack.push_async_callback(ws.close)
        print("✅ WebSocket 连接中...")

asyncio.run(main())

输出：

✅ WebSocket 连接中...
🔌 WebSocket 关闭连接

可以看到：即使这个 DummyWS 不支持 async with，我们仍然可以通过 push_async_callback 保证它在退出时自动调用 close()。

那么方法中可以带参数吗？ 当然可以！

from contextlib import AsyncExitStack
import asyncio

async def release(name):
    print(f"释放资源：{name}")


async def main():
    async with AsyncExitStack() as stack:
        stack.push_async_callback(release, "redis")
        stack.push_async_callback(release, "mysql")
        print("处理中...")


asyncio.run(main())

退出时将按注册顺序的逆序执行：

处理中...
释放资源：mysql
释放资源：redis