1. 从一次令人抓狂的资源泄漏说起大概两年前我接手过一个内部工具功能很简单定时拉数据、解析、入库。代码写得规矩但跑上几天就会报OSError: [Errno 24] Too many open files。当时我第一反应是调大文件描述符上限后来才发现问题的根子在于代码里到处是裸open()没有with兜底——资源用完不关系统迟早给你颜色看。1.1 资源泄漏的真实形态先看一段“看起来完全没问题”的代码def load_config(path): f open(path, r) data f.read() return json.loads(data)这段代码在很多环境里不会立刻出问题因为解释器自带的引用计数机制会在函数退出时让f被回收文件也随之关闭。但这里至少依赖三个不稳定的前提你的解释器实现确实能及时触发引用计数回收函数是正常返回而不是中途抛异常某些扩展模块没有拖慢内存回收节奏。任何一个前提不满足文件描述符就会一直挂着。更经典的翻车场景是在循环里开文件for path in paths: f open(path, r) process(f) # 如果 process 中途抛异常f 永远不被关闭这不是“新手才会犯的错误”我在不少维护多年的老项目里都见过。数据库游标、临时目录、分布式锁、网络连接本质上都是同一种资源有借有还。资源管理这件事和业务逻辑无关它是程序正确运行的地基。1.2 try/finally 为什么不够用很多人会说不关文件那我用 try/finally 不就行了f open(path, r) try: data f.read() finally: f.close()确实能解决“一定关闭”的问题但代码一多就露馅了。资源一多try/finally 会变成三层嵌套中途忘记写 finally 是家常便饭而且 close 本身还可能抛异常得再包一层。此时 Python 给出的标准答案是with语句。很多教程会把 with 解释成“自动关闭文件”这个总结太浅了。真正的问题是你知道 with 能保证什么、不能保证什么以及什么时候该自己写一个上下文管理器吗这篇文章就是把这些事一次说透。2. with 的底层协议enter和exit到底在配合什么with不是魔法它只是两个协议钩子的编排。任何对象只要实现了__enter__和__exit__两个方法就能被with使用。2.1 三步执行顺序与三个异常参数执行顺序是固定的先调用obj.__enter__()返回值赋给as后面的变量执行 with 代码块无论代码块是否抛异常都会调用obj.__exit__(exc_type, exc_val, exc_tb)。第三步里的三个参数是理解 with 的关键。没有异常时三个参数全是None有异常时分别对应异常类型、异常实例、traceback。自己实现一遍就全清楚了class ManagedFile: def __init__(self, path): self.path path def __enter__(self): self.f open(self.path, r) return self.f def __exit__(self, exc_type, exc_val, exc_tb): self.f.close() return False多个资源的 with 写法也值得注意资源从左到右依次进入如果中途某个__enter__抛异常前面已经进入的资源会按从右往左的顺序清理。这个顺序是解释器帮你兜底的不用你手动操心。2.2exit返回值的两种语义__exit__的返回值只有两档返回False或None异常继续向外抛返回True异常被抑制with 块外面的人看不到这个异常。把常见情况整理成一张表场景enter返回值代码块是否异常exit三个参数exit返回值异常最终去向正常执行资源对象否全为 None随意通常是 False/None无异常代码块抛异常资源对象是类型/实例/tracebackFalse 或 None继续向外传播故意吞掉异常资源对象是类型/实例/tracebackTrue被静默处理新手最容易犯的错误是在__exit__里打了日志、又没把异常重新 raise想当然地以为“我已经处理过了”。结果__exit__自然返回None异常照样往外冒。想吞异常必须显式return True。反过来滥用True也会掩盖 bug——该让调用方知道的错误被静默吞掉后续排查成本会翻几倍。3. contextlib 四件套能少写代码的地方绝不多写类实现能做到的事情标准库contextlib大部分能以更轻量的方式做到。我最常用的是四个contextmanager、closing、suppress、nullcontext。3.1 contextmanager 生成器大法contextmanager把生成器的 yield 切成两半yield 之前的代码等价于__enter__yield 之后的代码等价于__exit__。配合 try/finally 可以精确控制清理逻辑。举例写一个带日志的计时上下文import time from contextlib import contextmanager contextmanager def elapsed(label): start time.perf_counter() print(f[{label}] start) try: yield finally: cost time.perf_counter() - start print(f[{label}] finished in {cost:.3f}s)调用方保持干净with elapsed(parse): parse_all()为什么 yield 必须放在 try/finally 里因为如果 with 代码块里抛了异常而你的清理代码只是写在 yield 后面没有 finally 包裹异常会直接从生成器里抛出清理逻辑就被跳过了。这个坑我在代码评审里见到太多次。那么contextmanager什么时候不够用当你需要在清理阶段拿到异常类型做分支判断时生成器方案很别扭——因为你只能在 except 分支里记录很难优雅地基于异常状态改变外部行为。一旦出现这种需求直接写类。3.2 closing、suppress、nullcontextclosing专门对付那些只有close()方法、没有实现上下文协议的对象比如某些第三方库返回的会话对象from contextlib import closing with closing(create_session()) as s: s.query(...)suppress是“主动选择吞掉特定异常”的上下文比 try/except 短得多from contextlib import suppress with suppress(FileNotFoundError): os.remove(tmp_path)适合“本来就允许缺省”的操作但别滥用不是所有异常都该静默。nullcontext很多人不知道它的作用就是“什么都不做”常用于统一接口。比如某个函数可能需要加锁也可能不需要from contextlib import nullcontext lock maybe_get_lock() # 可能为 None cm lock if lock else nullcontext() with cm: do_something()这四件套覆盖了日常 80% 的需求。我自己写项目时能不用类就不用类生成器方案更直白、代码量少一半。4. 手写上下文管理器类方案和生成器方案怎么选业务里真正需要手写上下文管理器的场景集中在数据库事务、分布式锁、连接池借还、临时环境变量设置这几种。选型的核心判断是上下文是否需要跨 with 块保留状态以及是否需要对异常做分支响应。4.1 类方案适合有状态、可复用的资源数据库事务是典型场景因为要根据有无异常决定 commit 还是 rollbackclass Transaction: def __init__(self, conn): self.conn conn def __enter__(self): return self.conn def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: self.conn.commit() else: self.conn.rollback() return False注意这里不要返回True把异常继续抛给上层让调用方知道事务失败了。提交/回滚逻辑收进__exit__之后业务代码里就不需要到处写 try/except 了。4.2 生成器方案适合一次性、轻量的封装如果上下文不需要保存跨块状态用contextmanager更简洁。比如临时修改环境变量contextmanager def temporary_env(key, value): old os.environ.get(key) os.environ[key] value try: yield finally: if old is None: del os.environ[key] else: os.environ[key] old这个例子里唯一重要的就是 finally保证环境变量一定还原。生成器方案一眼能看出“进入时做什么、退出时做什么”代码评审成本低。4.3 两者的对比维度类实现contextmanager代码量偏多需要完整类结构少函数加装饰器即可访问异常参数直接拿到三个参数需要 try/except稍绕跨多个 with 保留状态方便对象属性可复用不方便生成器一般一次性复杂清理逻辑清晰方法内可拆分靠 finally 组织可读性更正式更直白提示contextmanager装饰的函数内部如果忘了写 yield运行时会直接抛RuntimeError。这种错误通常出现在重构清理逻辑、误删了 yield 行的时候排查时要先往这方面想。5. 事务、锁、连接池with 在真实业务里的正确姿势前四节解决“会不会写”这一节解决“会不会用对”。资源管理的正确姿势集中在三个关键词异常分支、超时兜底、借用与归还。5.1 事务提交/回滚的异常分支前面那个 Transaction 只处理了“有没有异常”。真实场景更复杂有些异常应当回滚有些异常回滚后仍需重新抛出有些甚至要先释放嵌套资源再回滚。遇到嵌套事务时我的做法是维护层级标记只有最外层才真正提交内层只标记“需要回滚”避免部分提交导致的脏数据。这属于业务层面的设计但上下文管理器正好提供了承载这种逻辑的位置。5.2 锁必须配超时threading.Lock对象原生支持with因为它内部实现了__enter__/__exit__。但最容易被忽略的是直接with lock:没有超时保护。如果某个线程拿锁后挂起其他线程会无限等待。正确的做法是if not lock.acquire(timeout3): logger.warning(lock acquire timeout) raise TimeoutError(lock timeout) try: do_critical() finally: lock.release()这里 try/finally 不可避免因为没有现成的上下文能表达“带超时的锁获取”和“获取失败时的分支”。你也可以自己封装一层contextmanager def timed_lock(lock, timeout): acquired lock.acquire(timeouttimeout) if not acquired: raise TimeoutError(facquire lock timeout: {timeout}s) try: yield finally: lock.release()封装之后调用方依然是干净的with timed_lock(lock, 3):超时兜底已经在内部完成。这是 with 最常见的扩展方式把资源处理逻辑塞进上下文业务代码保持在最外面。5.3 连接池的借与还数据库连接池里借出连接和归还连接天然是一对 enter/exitcontextmanager def acquire(pool): conn pool.get() try: yield conn except Exception: conn.mark_broken() raise else: pool.put(conn)这里有个细节连接出现异常时不能归还给池子否则下一批人拿到坏连接。这个分支逻辑放上下文里再合适不过。很多人借还连接只会在 finally 里put回去坏连接在池里反复造成间歇性故障排查起来相当痛苦。5.4 别忘了 async with异步代码也有对称的协议__aenter__、__aexit__原理与同步版一致只是清理逻辑往往也要等待 I/O。异步 HTTP 客户端、异步锁这些资源都应该显式用async with。异步代码里的资源泄漏更难受因为异常栈经常跨协程变量作用域也更隐蔽。我的原则很简单凡是异步资源一律走 async 上下文绝不裸持有。6. ExitStack把“不确定数量”的资源也管起来前面讨论的都是数量固定的资源但有时你根本不知道会开几个文件、加几把锁。这个场景里ExitStack是专门工具。6.1 进入几个就退出几个比如一次要处理 20 个文件中途第 15 个打不开前面 14 个也得清理。写成普通循环勉强能扛住但如果文件数量是动态的就麻烦了from contextlib import ExitStack with ExitStack() as stack: files [] for path in paths: f open(path, r) files.append(f) stack.enter_context(f) for f in files: process(f)enter_context会把每个文件注册进栈with 块结束或中途异常时按后进先出的顺序全部清理。动态数量资源的管理从笨拙的手动列表加 finally变成了一行 API。6.2 callback、push 与 pop_allExitStack 还能注册任意回调适合做非对象化的清理比如“退出时删除临时目录”“退出时发送心跳”。更实用的是pop_all你可以把当前栈里的清理责任整体交出去由更高层的上下文统一负责。在“函数内部临时注册资源、由调用方统一清理”的模式里非常顺手def prepare(stack: ExitStack): tmp stack.enter_context(temp_dir()) cfg stack.enter_context(open(config_path)) return cfg with ExitStack() as stack: cfg prepare(stack) # prepare 里注册的资源由外层的 stack 统一收起这种“把 ExitStack 传进函数”的模式比用全局变量管理资源干净得多。写插件或框架级代码时体验尤其明显。6.3 适合替换的征兆如果你发现代码里出现三个以上层层嵌套的 with或者一堆手动 close 排在 finally 里多半就该换成 ExitStack。嵌套 with 不是不可读但读起来得在脑子里维护一个资源栈ExitStack 把这种栈显式化谁注册的、谁负责、撤销顺序是什么一眼就能看明白。最后分享一个自己的惯性动作接到新代码时我会全局搜open(、.acquire(、pool.get(这几种调用凡是没跟在 with 或 ExitStack 后面的默认标成潜在泄漏点。这个检查动作救过我不少次。写出入口配对清楚、退出路径可控的代码写的时候多花五分钟运行的时候能少熬好几个夜。 
长河君 · 2026/10/20 18:06:02 · 阅读 1.2k Python上下文管理器全面解析：with、contextlib与ExitStack实战
1. 从一次令人抓狂的资源泄漏说起大概两年前我接手过一个内部工具功能很简单定时拉数据、解析、入库。代码写得规矩但跑上几天就会报OSError: [Errno 24] Too many open files。当时我第一反应是调大文件描述符上限后来才发现问题的根子在于代码里到处是裸open()没有with兜底——资源用完不关系统迟早给你颜色看。1.1 资源泄漏的真实形态先看一段“看起来完全没问题”的代码def load_config(path): f open(path, r) data f.read() return json.loads(data)这段代码在很多环境里不会立刻出问题因为解释器自带的引用计数机制会在函数退出时让f被回收文件也随之关闭。但这里至少依赖三个不稳定的前提你的解释器实现确实能及时触发引用计数回收函数是正常返回而不是中途抛异常某些扩展模块没有拖慢内存回收节奏。任何一个前提不满足文件描述符就会一直挂着。更经典的翻车场景是在循环里开文件for path in paths: f open(path, r) process(f) # 如果 process 中途抛异常f 永远不被关闭这不是“新手才会犯的错误”我在不少维护多年的老项目里都见过。数据库游标、临时目录、分布式锁、网络连接本质上都是同一种资源有借有还。资源管理这件事和业务逻辑无关它是程序正确运行的地基。1.2 try/finally 为什么不够用很多人会说不关文件那我用 try/finally 不就行了f open(path, r) try: data f.read() finally: f.close()确实能解决“一定关闭”的问题但代码一多就露馅了。资源一多try/finally 会变成三层嵌套中途忘记写 finally 是家常便饭而且 close 本身还可能抛异常得再包一层。此时 Python 给出的标准答案是with语句。很多教程会把 with 解释成“自动关闭文件”这个总结太浅了。真正的问题是你知道 with 能保证什么、不能保证什么以及什么时候该自己写一个上下文管理器吗这篇文章就是把这些事一次说透。2. with 的底层协议enter和exit到底在配合什么with不是魔法它只是两个协议钩子的编排。任何对象只要实现了__enter__和__exit__两个方法就能被with使用。2.1 三步执行顺序与三个异常参数执行顺序是固定的先调用obj.__enter__()返回值赋给as后面的变量执行 with 代码块无论代码块是否抛异常都会调用obj.__exit__(exc_type, exc_val, exc_tb)。第三步里的三个参数是理解 with 的关键。没有异常时三个参数全是None有异常时分别对应异常类型、异常实例、traceback。自己实现一遍就全清楚了class ManagedFile: def __init__(self, path): self.path path def __enter__(self): self.f open(self.path, r) return self.f def __exit__(self, exc_type, exc_val, exc_tb): self.f.close() return False多个资源的 with 写法也值得注意资源从左到右依次进入如果中途某个__enter__抛异常前面已经进入的资源会按从右往左的顺序清理。这个顺序是解释器帮你兜底的不用你手动操心。2.2exit返回值的两种语义__exit__的返回值只有两档返回False或None异常继续向外抛返回True异常被抑制with 块外面的人看不到这个异常。把常见情况整理成一张表场景enter返回值代码块是否异常exit三个参数exit返回值异常最终去向正常执行资源对象否全为 None随意通常是 False/None无异常代码块抛异常资源对象是类型/实例/tracebackFalse 或 None继续向外传播故意吞掉异常资源对象是类型/实例/tracebackTrue被静默处理新手最容易犯的错误是在__exit__里打了日志、又没把异常重新 raise想当然地以为“我已经处理过了”。结果__exit__自然返回None异常照样往外冒。想吞异常必须显式return True。反过来滥用True也会掩盖 bug——该让调用方知道的错误被静默吞掉后续排查成本会翻几倍。3. contextlib 四件套能少写代码的地方绝不多写类实现能做到的事情标准库contextlib大部分能以更轻量的方式做到。我最常用的是四个contextmanager、closing、suppress、nullcontext。3.1 contextmanager 生成器大法contextmanager把生成器的 yield 切成两半yield 之前的代码等价于__enter__yield 之后的代码等价于__exit__。配合 try/finally 可以精确控制清理逻辑。举例写一个带日志的计时上下文import time from contextlib import contextmanager contextmanager def elapsed(label): start time.perf_counter() print(f[{label}] start) try: yield finally: cost time.perf_counter() - start print(f[{label}] finished in {cost:.3f}s)调用方保持干净with elapsed(parse): parse_all()为什么 yield 必须放在 try/finally 里因为如果 with 代码块里抛了异常而你的清理代码只是写在 yield 后面没有 finally 包裹异常会直接从生成器里抛出清理逻辑就被跳过了。这个坑我在代码评审里见到太多次。那么contextmanager什么时候不够用当你需要在清理阶段拿到异常类型做分支判断时生成器方案很别扭——因为你只能在 except 分支里记录很难优雅地基于异常状态改变外部行为。一旦出现这种需求直接写类。3.2 closing、suppress、nullcontextclosing专门对付那些只有close()方法、没有实现上下文协议的对象比如某些第三方库返回的会话对象from contextlib import closing with closing(create_session()) as s: s.query(...)suppress是“主动选择吞掉特定异常”的上下文比 try/except 短得多from contextlib import suppress with suppress(FileNotFoundError): os.remove(tmp_path)适合“本来就允许缺省”的操作但别滥用不是所有异常都该静默。nullcontext很多人不知道它的作用就是“什么都不做”常用于统一接口。比如某个函数可能需要加锁也可能不需要from contextlib import nullcontext lock maybe_get_lock() # 可能为 None cm lock if lock else nullcontext() with cm: do_something()这四件套覆盖了日常 80% 的需求。我自己写项目时能不用类就不用类生成器方案更直白、代码量少一半。4. 手写上下文管理器类方案和生成器方案怎么选业务里真正需要手写上下文管理器的场景集中在数据库事务、分布式锁、连接池借还、临时环境变量设置这几种。选型的核心判断是上下文是否需要跨 with 块保留状态以及是否需要对异常做分支响应。4.1 类方案适合有状态、可复用的资源数据库事务是典型场景因为要根据有无异常决定 commit 还是 rollbackclass Transaction: def __init__(self, conn): self.conn conn def __enter__(self): return self.conn def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: self.conn.commit() else: self.conn.rollback() return False注意这里不要返回True把异常继续抛给上层让调用方知道事务失败了。提交/回滚逻辑收进__exit__之后业务代码里就不需要到处写 try/except 了。4.2 生成器方案适合一次性、轻量的封装如果上下文不需要保存跨块状态用contextmanager更简洁。比如临时修改环境变量contextmanager def temporary_env(key, value): old os.environ.get(key) os.environ[key] value try: yield finally: if old is None: del os.environ[key] else: os.environ[key] old这个例子里唯一重要的就是 finally保证环境变量一定还原。生成器方案一眼能看出“进入时做什么、退出时做什么”代码评审成本低。4.3 两者的对比维度类实现contextmanager代码量偏多需要完整类结构少函数加装饰器即可访问异常参数直接拿到三个参数需要 try/except稍绕跨多个 with 保留状态方便对象属性可复用不方便生成器一般一次性复杂清理逻辑清晰方法内可拆分靠 finally 组织可读性更正式更直白提示contextmanager装饰的函数内部如果忘了写 yield运行时会直接抛RuntimeError。这种错误通常出现在重构清理逻辑、误删了 yield 行的时候排查时要先往这方面想。5. 事务、锁、连接池with 在真实业务里的正确姿势前四节解决“会不会写”这一节解决“会不会用对”。资源管理的正确姿势集中在三个关键词异常分支、超时兜底、借用与归还。5.1 事务提交/回滚的异常分支前面那个 Transaction 只处理了“有没有异常”。真实场景更复杂有些异常应当回滚有些异常回滚后仍需重新抛出有些甚至要先释放嵌套资源再回滚。遇到嵌套事务时我的做法是维护层级标记只有最外层才真正提交内层只标记“需要回滚”避免部分提交导致的脏数据。这属于业务层面的设计但上下文管理器正好提供了承载这种逻辑的位置。5.2 锁必须配超时threading.Lock对象原生支持with因为它内部实现了__enter__/__exit__。但最容易被忽略的是直接with lock:没有超时保护。如果某个线程拿锁后挂起其他线程会无限等待。正确的做法是if not lock.acquire(timeout3): logger.warning(lock acquire timeout) raise TimeoutError(lock timeout) try: do_critical() finally: lock.release()这里 try/finally 不可避免因为没有现成的上下文能表达“带超时的锁获取”和“获取失败时的分支”。你也可以自己封装一层contextmanager def timed_lock(lock, timeout): acquired lock.acquire(timeouttimeout) if not acquired: raise TimeoutError(facquire lock timeout: {timeout}s) try: yield finally: lock.release()封装之后调用方依然是干净的with timed_lock(lock, 3):超时兜底已经在内部完成。这是 with 最常见的扩展方式把资源处理逻辑塞进上下文业务代码保持在最外面。5.3 连接池的借与还数据库连接池里借出连接和归还连接天然是一对 enter/exitcontextmanager def acquire(pool): conn pool.get() try: yield conn except Exception: conn.mark_broken() raise else: pool.put(conn)这里有个细节连接出现异常时不能归还给池子否则下一批人拿到坏连接。这个分支逻辑放上下文里再合适不过。很多人借还连接只会在 finally 里put回去坏连接在池里反复造成间歇性故障排查起来相当痛苦。5.4 别忘了 async with异步代码也有对称的协议__aenter__、__aexit__原理与同步版一致只是清理逻辑往往也要等待 I/O。异步 HTTP 客户端、异步锁这些资源都应该显式用async with。异步代码里的资源泄漏更难受因为异常栈经常跨协程变量作用域也更隐蔽。我的原则很简单凡是异步资源一律走 async 上下文绝不裸持有。6. ExitStack把“不确定数量”的资源也管起来前面讨论的都是数量固定的资源但有时你根本不知道会开几个文件、加几把锁。这个场景里ExitStack是专门工具。6.1 进入几个就退出几个比如一次要处理 20 个文件中途第 15 个打不开前面 14 个也得清理。写成普通循环勉强能扛住但如果文件数量是动态的就麻烦了from contextlib import ExitStack with ExitStack() as stack: files [] for path in paths: f open(path, r) files.append(f) stack.enter_context(f) for f in files: process(f)enter_context会把每个文件注册进栈with 块结束或中途异常时按后进先出的顺序全部清理。动态数量资源的管理从笨拙的手动列表加 finally变成了一行 API。6.2 callback、push 与 pop_allExitStack 还能注册任意回调适合做非对象化的清理比如“退出时删除临时目录”“退出时发送心跳”。更实用的是pop_all你可以把当前栈里的清理责任整体交出去由更高层的上下文统一负责。在“函数内部临时注册资源、由调用方统一清理”的模式里非常顺手def prepare(stack: ExitStack): tmp stack.enter_context(temp_dir()) cfg stack.enter_context(open(config_path)) return cfg with ExitStack() as stack: cfg prepare(stack) # prepare 里注册的资源由外层的 stack 统一收起这种“把 ExitStack 传进函数”的模式比用全局变量管理资源干净得多。写插件或框架级代码时体验尤其明显。6.3 适合替换的征兆如果你发现代码里出现三个以上层层嵌套的 with或者一堆手动 close 排在 finally 里多半就该换成 ExitStack。嵌套 with 不是不可读但读起来得在脑子里维护一个资源栈ExitStack 把这种栈显式化谁注册的、谁负责、撤销顺序是什么一眼就能看明白。最后分享一个自己的惯性动作接到新代码时我会全局搜open(、.acquire(、pool.get(这几种调用凡是没跟在 with 或 ExitStack 后面的默认标成潜在泄漏点。这个检查动作救过我不少次。写出入口配对清楚、退出路径可控的代码写的时候多花五分钟运行的时候能少熬好几个夜。