导读:本期聚焦于赵景明创作的《Python 异步上下文管理器怎么实现?async with 的正确用法与实战示例》,敬请观看详情。Python异步编程中经常需要管理数据库连接、网络请求、文件读写等需要异步释放的资源,这时同步的with语句就无能为力了。本文围绕异步上下文管理器展开,详细讲解如何通过定义__aenter__和__aexit__两个魔法方法来自定义异步上下文管理器,以及如何使用contextlib提供的asynccontextmanager装饰器快速实现。文章还介绍了AsyncExitStack管理多个异步资源的技巧、常见异常处理陷阱、超时控制方案,并通过数据库连接池和HTTP客户端的实际案例演示完整用法,帮助你写出安全、优雅、可维护的异步Python代码。

在编写异步程序时,资源的获取与释放同样需要遵守“用了就要还”的原则。数据库连接、网络会话、异步锁、临时文件,这些资源如果只获取不释放,轻则造成内存泄漏,重则导致连接池耗尽、程序卡死。同步世界里我们用 with 语句配合上下文管理器解决这个问题,而在 asyncio 的世界里,普通 with 语句无法等待异步操作完成,必须依赖异步上下文管理器,也就是配合 async with 语句使用的对象。本文将从原理到实战,完整讲解异步上下文管理器的几种实现方式。

Python 异步上下文管理器怎么实现?async with 的正确用法与实战示例

异步上下文管理器的底层原理:__aenter__ 与 __aexit__

要理解异步上下文管理器,可以先回顾同步版本的工作机制。同步上下文管理器定义了 __enter____exit__ 两个方法,当解释器执行 with 语句时,先调用 __enter__ 获取资源,代码块执行完毕后无论是否抛出异常,都会调用 __exit__ 做清理。如果 __exit__ 返回 True,异常会被吞掉;返回 False 或 None,异常继续向外传播。

异步版本把这个协议“异步化”了:对应的方法变成了 __aenter____aexit__,它们都是协程函数,内部可以执行 await 表达式。当 async with 语句执行时,事件循环会 await 这两个方法。这个设计解决了一个核心问题——资源的获取和释放本身就是耗时操作。比如打开一个数据库连接需要网络往返,释放连接时需要发送断开指令,这些都必须以非阻塞的方式完成,否则会拖垮整个事件循环。

下面用纯手工方式实现一个异步上下文管理器,模拟一个异步数据库连接:

import asyncio

class AsyncConnection:
    def __init__(self, host):
        self.host = host
        self.connected = False

    async def __aenter__(self):
        # 模拟异步建立连接的网络开销
        await asyncio.sleep(0.1)
        self.connected = True
        print(f"已连接到 {self.host}")
        return self  # 返回值会赋给 as 后面的变量

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 无论代码块是否抛异常,这里都会执行
        await asyncio.sleep(0.1)
        self.connected = False
        print(f"已断开 {self.host} 的连接")
        if exc_type is not None:
            print(f"代码块中发生了异常: {exc_val}")
        return False  # 不吞掉异常,继续向外抛

async def main():
    async with AsyncConnection("db.ipipp.com") as conn:
        print("正在执行查询...")
        await asyncio.sleep(0.2)

asyncio.run(main())

运行这段代码可以看到,即使把代码块里的 await asyncio.sleep(0.2) 换成一行会抛异常的语句,__aexit__ 依然会被执行,连接依然会被正确释放。__aexit__ 的三个参数分别对应异常类型、异常实例和 traceback 对象,正常结束时三者都是 None。手动实现的好处是逻辑清晰、可以持有状态,适合封装成可复用的类,缺点是代码量偏多。

轻量方案:用 asynccontextmanager 装饰器快速封装

如果只是临时使用某个异步资源,写一个完整的类显得很重。标准库 contextlib 提供了 asynccontextmanager 装饰器,只需把一个异步生成器函数装饰一下,就能得到现成的异步上下文管理器。yield 之前的代码相当于 __aenter__,yield 之后的代码相当于 __aexit__,写起来和 try-finally 非常接近。

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def open_connection(host):
    conn = await create_connection(host)  # 进入时获取资源
    try:
        yield conn  # yield 的值就是 as 后面拿到的对象
    finally:
        await conn.close()  # 退出时释放资源,即使异常也会执行

async def create_connection(host):
    await asyncio.sleep(0.1)
    print(f"连接 {host} 成功")
    return SimpleNamespaceConn(host)

class SimpleNamespaceConn:
    def __init__(self, host):
        self.host = host
    async def close(self):
        await asyncio.sleep(0.1)
        print(f"关闭 {self.host}")

async def main():
    async with open_connection("api.ipipp.com") as conn:
        print(f"使用连接访问 {conn.host}")

asyncio.run(main())

这种写法有几个值得注意的细节。第一,异步生成器函数只能 yield 一次,多次 yield 会抛出 RuntimeError。第二,如果希望在异常发生时做特殊处理,可以在 yield 外面套一层 try-except,捕获到异常后可以选择记录日志再重新抛出,或者直接吞掉异常。第三,被装饰后的对象是可重入性受限的,同一个装饰好的函数每次调用都会产生新的上下文管理器实例,这一点和类方式一致,通常不构成问题。

对比两种实现方式:类方式适合需要复用、携带复杂状态、支持多次进入的场景,比如连接池对象;装饰器方式适合一次性封装、逻辑简单的场景,比如临时打开一个文件或者测量某段异步代码的耗时。实际项目中两者经常混用,先用类封装核心资源,再用装饰器包装常用组合。

进阶技巧:AsyncExitStack 管理多个异步资源与异常陷阱

当需要同时管理多个异步资源时,层层嵌套 async with 会让代码缩进越来越深,可读性急剧下降。contextlib 提供的 AsyncExitStack 可以把注册资源获取和实际使用分离开,用循环或条件语句动态管理任意数量的异步上下文:

import asyncio
from contextlib import AsyncExitStack

async def main():
    async with AsyncExitStack() as stack:
        # 动态注册多个连接,退出时按注册的逆序自动释放
        conns = [
            await stack.enter_async_context(make_conn(i))
            for i in range(3)
        ]
        for conn in conns:
            await conn.query()

async def make_conn(i):
    from contextlib import asynccontextmanager

    @asynccontextmanager
    async def _cm():
        print(f"打开连接 {i}")
        try:
            yield {"id": i}
        finally:
            print(f"关闭连接 {i}")
    return _cm()

asyncio.run(main())

AsyncExitStackenter_async_context 方法接收一个异步上下文管理器,立即执行进入逻辑,并把退出逻辑压入栈中。整个 stack 退出时按照后进先出的顺序逐个调用清理函数,这和嵌套 with 的行为完全一致。它还支持通过 callback 注册任意异步回调,灵活性更高。

使用异步上下文管理器时还有几个常见陷阱需要留意。其一是不要在 __aexit__ 或 finally 块中写出可能永久阻塞的代码,比如 await 一个永远不会完成的任务,这会导致资源永远无法释放。其二是异常吞噬问题,__aexit__ 返回 True 会静默吞掉异常,如果不熟悉这个语义,排查 bug 会非常痛苦,建议默认返回 False 或 None。其三是超时控制,可以结合 asyncio.timeoutasyncio.wait_for 把整个 async with 块包起来,防止获取资源的步骤卡死:

import asyncio

async def main():
    try:
        async with asyncio.timeout(2.0):  # 整块代码限时2秒
            async with get_slow_connection() as conn:
                await conn.query()
    except TimeoutError:
        print("获取连接超时,已自动退出上下文")

总结一下,异步上下文管理器是异步 Python 中保证资源安全的基石。掌握类方式实现 __aenter____aexit__、用 asynccontextmanager 快速封装简单场景、用 AsyncExitStack 应对动态多资源的组合,再配合超时控制和对异常语义的清晰理解,就能写出既优雅又健壮的异步代码。建议在自己的项目中把所有异步资源的获取释放统一收敛到这三套模式里,代码的可维护性会有明显提升。

Python异步编程异步上下文管理器async with修改时间:2026-09-12 09:11:16

免责声明:已尽一切努力确保本网站所含信息的准确性。网站作品多为原创整理与精心创作,观点力求客观中立。本站旨在免费分享,内容仅供个人学习、研究或参考使用。若引用了第三方作品,版权归原作者所有。如内容涉及您的权益,请联系我们进行处理Email:chomcom@qq.com。
引用或转载本作品时,请注明当前出处:https://www.ipipp.com/html/20260912/55235.html,基于非商业用途的前提下,欢迎转载或二创本作品。
内容垂直聚焦
专注技术核心技术栏目,确保每篇文章深度聚焦于实用技能。从代码技巧到架构设计,为用户提供无干扰的纯技术知识沉淀,精准满足专业提升需求。
知识结构清晰
覆盖从开发到部署的全链路。AI、前端、编程、数据库、服务器、建站、系统层层递进,构建清晰学习路径,帮助用户系统化掌握开发与运维所需的核心技术。
深度技术解析
拒绝泛泛而谈,深入技术细节与实践难点。无论是数据库优化还是服务器配置,均结合真实场景与代码示例进行剖析,致力于提供可直接应用于工作的解决方案。
专业领域覆盖
精准对应开发生命周期。从前端界面到后端编程,从数据库操作到服务器运维,形成完整闭环,一站式满足全栈工程师和运维人员的技术需求。
即学即用高效
内容强调实操性,步骤清晰、代码完整。用户可根据教程直接复现和应用于自身项目,显著缩短从学习到实践的距离,快速解决开发中的具体问题。
持续更新保障
专注既定技术方向进行长期、稳定的内容输出。确保各栏目技术文章持续更新迭代,紧跟主流技术发展趋势,为用户提供经久不衰的学习价值。