掌握Python异步上下文管理器（深入浅出async with用法与实战）

什么是异步上下文管理器？

普通上下文管理器（使用 with 语句）大家可能比较熟悉，比如打开文件：

with open('file.txt', 'r') as f:    content = f.read()# 文件自动关闭

但在异步编程中（使用 async/await），如果资源操作本身是异步的（比如数据库连接、HTTP请求），就需要异步上下文管理器，它使用 async with 语法，并实现 __aenter__ 和 __aexit__ 方法。

如何定义一个异步上下文管理器？

你可以通过类或使用 contextlib.asynccontextmanager 装饰器来创建。下面我们分别演示。

方法一：使用类定义

import asyncioclass AsyncDatabaseConnection:    def __init__(self, db_name):        self.db_name = db_name    async def __aenter__(self):        print(f"正在连接数据库 {self.db_name}...")        await asyncio.sleep(1)  # 模拟异步连接耗时        self.connection = f"Connection to {self.db_name}"        return self.connection    async def __aexit__(self, exc_type, exc_val, exc_tb):        print(f"正在关闭数据库 {self.db_name} 的连接...")        await asyncio.sleep(0.5)  # 模拟异步关闭耗时        self.connection = None# 使用示例async def main():    async with AsyncDatabaseConnection("mydb") as conn:        print(f"已获取连接: {conn}")        # 在这里执行数据库操作# 运行asyncio.run(main())

方法二：使用 @asynccontextmanager 装饰器（推荐）

这种方式更简洁，适合快速创建简单的异步上下文管理器。

import asynciofrom contextlib import asynccontextmanager@asynccontextmanagerasync def async_file_open(filename):    print(f"正在异步打开文件 {filename}...")    await asyncio.sleep(0.2)    file_handle = f"FakeFileHandle({filename})"    try:        yield file_handle    finally:        print(f"正在异步关闭文件 {filename}...")        await asyncio.sleep(0.1)async def read_data():    async with async_file_open("data.txt") as f:        print(f"读取数据使用: {f}")asyncio.run(read_data())

为什么需要异步上下文管理器？

在异步编程中，很多 I/O 操作（如网络请求、数据库交互）都是非阻塞的，必须使用 await。如果仍用普通 with，就无法正确等待这些异步操作完成，可能导致资源未正确初始化或释放。

使用 异步资源管理 可以确保：

• 进入上下文时，异步初始化完成后再执行业务逻辑
• 退出上下文时，无论是否发生异常，都能异步清理资源
• 代码结构清晰，避免手动调用 close() 等方法

实际应用场景

以下是一些常见的使用场景：

• 异步数据库连接池（如 asyncpg、aiomysql）
• HTTP 客户端会话（如 aiohttp.ClientSession）
• 异步文件或网络流处理
• 分布式锁或限流器的异步获取与释放

例如，使用 aiohttp 发起多个异步请求：

import aiohttpimport asyncioasync def fetch(url):    async with aiohttp.ClientSession() as session:        async with session.get(url) as response:            return await response.text()async def main():    html = await fetch('https://httpbin.org/get')    print(len(html))asyncio.run(main())

总结

通过本文，你已经了解了 Python async 编程中异步上下文管理器的核心概念和使用方法。记住：

• 使用 async with 配合实现了 __aenter__ 和 __aexit__ 的对象
• 优先考虑使用 @asynccontextmanager 装饰器简化代码
• 它能确保异步资源的安全获取与释放，提升程序健壮性

掌握 Python异步上下文管理器 是迈向高级异步编程的重要一步。赶快动手试试吧！