
凌晨三点接到运维同事电话的那一刻我就知道自己迟早得写一篇关于Python上下文管理器的文章。那次事故现在想起来还心有余悸一个跑了好几个月的定时脚本突然把服务器的文件句柄耗光了进程崩溃磁盘里还留着几个没写完的半截日志。定位到最后就是一行代码——用open()打开了文件却因为中间抛了一次异常导致close()根本没机会执行。从那以后凡是获取资源的地方必须用with几乎成了我审查代码的第一条铁律。这篇文章我会把Python上下文管理器with语句的底层原理、实现方式和真实项目里的玩法一次性讲透适合已经会写Python但还没有深入理解with内部机制的开发者也适合刚入门、想从一开始就建立正确资源管理习惯的新手。1. 一个典型的资源泄漏事故为什么需要with语句1.1 从一段看起来没问题的代码讲起很多刚从其他语言转过来的同学都有过这样的经历写了几行打开文件的代码本地跑了一遍没问题就以为完事了。比如data open(config.json, r).read()这行代码吐出的JSON数据确实能解析但如果你开一个监控工具去看进程的文件描述符就会发现每次执行这行代码文件描述符数量就1而且是只增不减。更隐蔽的是读文件一般还好如果是写文件缓冲区里的数据可能还没来得及落盘程序就异常退出了最后你得到的是一个空文件或者残缺文件。我实际排查脚本故障时见过太多类似情况sqlite3.connect()建了连接但没调close()threading.Lock()加了锁但在某个分支里忘了release()还有往系统PATH里临时添加路径用完之后忘了恢复。这些问题本质上都指向同一件事——资源生命周期管理。操作系统给每个进程的文件描述符数量是有限制的通常几千到几万数据库连接池也是稀缺资源锁不释放就直接让别的线程死在等待里。1.2 传统的try/finally写法与它的痛点在with出现之前Python里规整的写法是try/finallyf open(config.json, r) try: data f.read() finally: f.close()这段代码比之前那行好得多无论read()是否抛异常close()都会执行。但问题也很明显——你只是读一个文件就要写三行样板代码。如果是多个资源叠加就成了套娃f1 open(a.txt, r) try: f2 open(b.txt, r) try: process(f1, f2) finally: f2.close() finally: f1.close()这种写法有几个实际痛点第一缩进层级越来越深代码可读性直线下降第二如果你忘了某一层finally资源泄漏就又回来了第三try块里如果还有业务逻辑很容易把资源释放和业务处理搞混。后来我参与新项目时就直接定了个规矩新人提交的代码里如果出现裸的open()且不在with里review直接打回。1.3 with语句的本质封装获取与释放两个动作with语句解决的就是这个获取资源、使用资源、释放资源三段式结构。它把资源的获取和释放动作封装成一个对象你只需要用with把它包起来剩下的交给Python解释器去保证with open(config.json, r) as f: data f.read()读到这里你可能会想这不就是语法糖吗对它确实是语法糖但这个语法糖背后有完整的协议支撑。理解了这套协议你不仅能放心地用内置的with open(...)还能自己写出各种花式上下文管理器让资源管理这件事变得非常优雅。下一节我就把这些糖纸拆开看看里面到底怎么运作。2. 协议层面拆解__enter__与__exit__的真实执行顺序2.1 上下文管理器协议的基本概念在Python里只要一个对象实现了__enter__(self)和__exit__(self, exc_type, exc_val, exc_tb)两个方法它就可以被with使用这个对象就叫上下文管理器。协议听起来很玄其实就是制定了游戏规则——解释器在遇到with语句时会按照固定顺序调用这两个方法。我用一个最简单的手写类演示一次class Demo: def __enter__(self): print(进入 with 块) return hello def __exit__(self, exc_type, exc_val, exc_tb): print(退出 with 块) return False with Demo() as value: print(value , value)执行这段代码输出顺序是进入 with 块 value hello 退出 with 块就这么简单。__enter__在with块开始前被调用它的返回值会赋给as后面的变量__exit__在with块结束后被调用负责清理工作。2.2 with语句背后那六步执行流程只看上面的输出还不够因为异常场景还没涉及。其实with语句完整执行流程是这样的先执行with后面的表达式拿到一个上下文管理器对象。解释器会立刻把它的__exit__方法保存起来这一点很关键——即使后面调用__enter__时出了问题也能确保__exit__有机会被调用。然后调用__enter__方法。如果写了as target__enter__的返回值会被赋值给target。执行with块内的代码块。无论with块内代码是否抛异常解释器都会调用之前保存的__exit__(exc_type, exc_val, exc_tb)。如果没异常三个参数全是None如果有异常三个参数分别是被抛异常的类型、异常实例、traceback对象。下面这段代码能让你直观地看到异常参数长什么样class ExceptionInspector: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): print(exc_type:, exc_type) print(exc_val:, exc_val) print(exc_tb:, exc_tb) return False try: with ExceptionInspector(): raise ValueError(出错了) except ValueError as e: print(捕获到异常:, e)输出exc_type: class ValueError exc_val: 出错了 exc_tb: traceback object at 0x... 捕获到异常: 出错了2.3 __exit__的返回值如何决定异常命运__exit__方法的返回值不是随便写写的。如果它返回True就表示我已经处理掉这个异常了解释器你不用再管了with块里的异常会被静默吞掉返回None或False异常会继续往外层抛和你不用with时的行为一致。这里有个新手常踩的坑__exit__里明明写了清理代码但忘了加return False或者写成了return True结果要么异常被意外吞掉要么什么都没清理。我建议你把它当成一个要不要压制异常的开关来理解class IgnoreError: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): return True # 不管发生什么都吞掉 with IgnoreError(): print(1 / 0) print(程序还在继续因为异常被吞了)这种什么都不管直接吞的写法一般不建议用会掩盖真实错误。但有些场景确实需要比如某些重试机制里的临时失败你可以在__exit__里判断异常类型来决定是否压制。后面讲contextlib时会有更优雅的替代方案。2.4 一个经常被忽略的细节as变量不一定非要有with open(...) as f是大家最熟悉的写法但不是所有上下文管理器都必须返回as变量。比如threading.Lock对象本身实现了协议你直接写with lock:就行它的__enter__返回的是锁对象自身就算你不赋值锁的获取和释放照样发生。还有一点容易混淆as后面的变量只在with块内有效。从块出来后这个变量名还存在但它的值往往已经处于已释放状态。最典型的就是文件with open(a.txt, r) as f: content f.read() print(f.closed) # True虽然变量f还能引用但它指向的文件对象已经被关闭了。3. 内置上下文管理器实测文件、锁与事务中的常见用法3.1 文件读写最基础的资源管理文件操作是with最常见的应用场景没有之一。我工作里写文件时几乎永远用下面这种形式with open(report.txt, w, encodingutf-8) as f: f.write(hello\n) f.write(world\n)open()返回的就是一个上下文管理器__exit__会帮你把文件关闭。这里有个我踩过的坑写中文文本时如果不指定encodingutf-8Windows下默认可能是gbk很容易抛UnicodeEncodeError。所以只要涉及文本文件我建议你永远显式指定编码。还有一个细节值得注意文件对象的__exit__除了关闭文件还会自动flush缓冲区。这意味着即使程序在with块内部崩了已经写入的数据也不会全丢之前缓冲的部分数据能被完整落盘。这也是为什么我一直强调能用with就别裸open。3.2 线程锁防止死锁的保险丝多线程编程里threading.Lock实现了上下文管理器接口with lock:等价于lock.acquire()和lock.release()的完美搭配import threading lock threading.Lock() shared_counter 0 def increment(): global shared_counter for _ in range(1000): with lock: shared_counter 1为什么这比手动acquire/release安全因为如果你在临界区里某个分支提前return或者抛了异常手动写法很可能漏掉release()导致死锁而with能在任何情况下释放锁相当于给临界区上下了一道保险丝。凡是需要try/finally来做资源释放的地方都是with的强项。3.3 数据库事务一个容易误用的连接对象数据库连接这块有个非常经典、也是面试高频的坑。很多人以为sqlite3.connect()返回的连接对象可以直接用于with管理连接关闭其实不然import sqlite3 conn sqlite3.connect(test.db) with conn: conn.execute(CREATE TABLE user (id INTEGER PRIMARY KEY, name TEXT)) conn.execute(INSERT INTO user (name) VALUES (?), (Alice,)) # with 结束时提交事务但连接并未关闭 print(conn) # sqlite3.Connection object at 0x...sqlite3.Connection的__exit__行为是如果with块内没有异常就COMMIT事务如果抛了异常就ROLLBACK事务。它根本不会调用close()所以严格来说它管理的是事务不是连接本身。如果你把with conn:当成用完自动关闭连接那程序运行久了连接数就会涨。真正要管理连接生命周期还是得with contextlib.closing(conn):或者干脆把连接和事务一起放在nullcontext里处理。不过事务场景确实很实用。比如写入一批数据要么全部成功要么全部回滚try: with conn: conn.execute(INSERT INTO user (name) VALUES (?), (Bob,)) conn.execute(INSERT INTO user (name) VALUES (?), (Carol,)) except sqlite3.IntegrityError: print(事务已回滚不会留下半截数据)3.4 临时修改全局配置decimal.localcontext除了文件和锁上下文管理器还能用来做临时修改、用完恢复的活。最典型的就是decimal模块的本地上下文import decimal with decimal.localcontext() as ctx: ctx.prec 50 result decimal.Decimal(1) / decimal.Decimal(7) print(result) # 50位精度 # 出了with块全局精度恢复成默认的28位 print(decimal.Decimal(1) / decimal.Decimal(7))这个场景特别适合做科学计算或者财务计算里的精度隔离。类似的还有临时改cwd当前工作目录之类的操作都可以用上下文管理器封装起来避免污染全局环境。理解了这种临时修改—自动还原的思维你会发现with能管的事情远比想象的多。4. 手写上下文管理器基于类与contextlib两条路线4.1 基于类实现完整、可控、可读性强当你要复用的资源管理逻辑比较复杂时用类来实现上下文管理器是最稳的。它最大的好处是__enter__和__exit__是两个独立的命名方法你可以在里面放很多辅助逻辑。比如我写过一个用于数据库备份的上下文管理器它负责建立连接、在退出时无论结果如何都关闭连接并且记录日志import logging class DBConnection: def __init__(self, db_name): self.db_name db_name self.conn None def __enter__(self): logging.info(f正在连接数据库: {self.db_name}) self.conn sqlite3.connect(self.db_name) return self.conn def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None: logging.error(f数据库操作异常: {exc_val}) self.conn.rollback() else: self.conn.commit() self.conn.close() logging.info(数据库连接已关闭) return False用的时候with DBConnection(app.db) as conn: conn.execute(UPDATE user SET name ? WHERE id ?, (Tester, 1))这比到处写connect/try/finally/close干净太多。并且如果你想在__exit__里决定是否吞掉异常只需要根据exc_type做分支返回True即可。比如某些已知的退出码错误可以当成正常结果这在很多第三方接口调用里很有用。4.2 基于contextlib.contextmanager用yield精简代码如果只是需要一个简单的资源管理片段写一整个类可能有点重。Python标准库给了我们一个更轻量级的方案——contextmanager装饰器。它的核心思想是把函数里yield之前的部分当作__enter__yield之后的部分当作__exit__。例如from contextlib import contextmanager contextmanager def open_managed(filename, mode): f open(filename, mode) try: yield f finally: f.close() with open_managed(test.txt, w) as f: f.write(hello)这个写法要比类短很多而且逻辑一目了然。特别要注意yield外面套了个try/finally这是必须的如果with块内部抛异常yield之后不会自动走下去必须靠finally确保清理代码执行。4.3 两种实现路线的选择原则不是无脑推荐哪一种我根据实际项目经验总结了三种对比维度对比维度基于类实现基于contextmanager实现代码量需要写完整方法代码量大函数内一条yield分隔很精简异常处理可以通过__exit__参数精确控制用try/except/finally在yield周围控制复杂状态实例属性天然适合保存状态需要借助闭包变量略麻烦复用性可以继承、扩展不能直接继承但可以用闭包封装适合场景复杂资源、需要精细管理、需要复用临时封装一段逻辑、代码量少我在项目中一般是这么定的超过20行且要在多个地方复用的资源管理就写类一次性使用的小逻辑用contextmanager。比如这个计时器用contextmanager写就很舒服import time from contextlib import contextmanager contextmanager def timing(label): start time.perf_counter() try: yield finally: elapsed time.perf_counter() - start print(f{label} 耗时: {elapsed:.4f}s) with timing(批量写入): time.sleep(0.2)4.4 contextmanager写法的隐性陷阱contextmanager虽然方便但有个坑大家容易踩yield的值其实就是as变量拿到的内容。如果你想让with块内部使用的正是资源本身记得yield资源对象如果只想表达一个进入-退出的时序不必写yield可以yield None。还有一个更隐蔽的问题yield前面的代码如果抛了异常后面的清理代码是否执行答案是不会——因为此时还没进入try。所以最好把资源获取放在try之前清理放在finally里这样能保证一旦获取成功必然有清理。5. 进阶组合拳ExitStack、嵌套与异步上下文管理5.1 多个资源并列用ExitStack避免嵌套地狱当你要同时打开好几个文件旧式写法很容易变成套娃with open(1.txt) as f1: with open(2.txt) as f2: with open(3.txt) as f3: pass这种代码确实能工作但缩进层级感人而且如果文件数量是动态的比如列表里10个文件你根本没法写死嵌套。contextlib.ExitStack就是为了解决这个问题诞生的。from contextlib import ExitStack filenames [1.txt, 2.txt, 3.txt] with ExitStack() as stack: files [stack.enter_context(open(name)) for name in filenames] for f in files: print(f.read())ExitStack允许你动态地把任意多个上下文管理器注册到它内部等到with ExitStack():块结束时所有注册过的上下文管理器会按照先入后出的顺序自动执行__exit__。就算在向列表中添加文件时中途抛异常已经被注册打开的那些文件也会被正确关闭。这个工具在处理不确定数量的外部资源时简直是救星。5.2 except/suppress/redirect_stdout标准库自带的一些神兵除了ExitStackcontextlib里还有几个我特别常用的工具。第一个是suppress它在知道可能抛某种异常但希望忽略时超级好用from contextlib import suppress import os with suppress(FileNotFoundError): os.remove(temp.txt) # 不用写try/except FileNotFoundError: pass第二个是redirect_stdout在测试代码或封装命令行工具时能临时接管print的输出from contextlib import redirect_stdout import io buffer io.StringIO() with redirect_stdout(buffer): print(这段内容不会出现在屏幕上) print(被重定向的内容:, buffer.getvalue())这些工具本质上都是上下文管理器读源码会发现它们要么用类实现要么用contextmanager实现。所以只要你吃透了协议标准库的很多高级能力都是顺理成章的。5.3 嵌套时的执行顺序谁先进入谁后退出多个with写在一行时执行顺序遵循先进后出类似栈class Tracer: def __init__(self, name): self.name name def __enter__(self): print(f进入 {self.name}) return self def __exit__(self, *args): print(f退出 {self.name}) return False with Tracer(A), Tracer(B): print(执行中)输出是进入 A 进入 B 执行中 退出 B 退出 A这个顺序在上层资源依赖下层资源的场景里尤其重要。比如先创建一个临时目录再在里面创建文件退出时就要先关文件再删临时目录正好符合先进后出。5.4 async with异步世界的上下文管理Python 3.5 之后异步编程也有一等公民的上下文管理器async with。它对应的协议是__aenter__和__aexit__这两个方法都必须是异步函数返回一个可等待对象。比如aiohttp的会话管理import aiohttp import asyncio async def fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text() asyncio.run(fetch(https://example.com))这两层async with分别负责关闭会话和响应避免手动写try/finally带来的异步资源泄漏。如果你自己写异步资源实现协议时记得两个方法都用async def并且通过return False决定异常传递。异步上下文管理器还有一个坑__aenter__和__aexit__内部不能调用time.sleep()这类同步阻塞函数否则会阻塞整个事件循环。需要等待时用await asyncio.sleep()。5.5 最后的经验先想清楚资源边界再写with写到最后分享一个我自己的实践经验。刚开始学with时我只会把它套在文件操作上后来项目里遇到各种奇怪的资源泄漏才慢慢养成一个习惯——每当开始写一段获取某样东西、用完需要释放的逻辑时先停下来想一下这个东西的生命周期边界在哪然后直接把这个边界用with圈出来。你可能会问with能管事务、管锁、管文件那全项目所有资源都这么写会不会过度我的经验是只要做的是获取-使用-释放三段式哪怕只是临时修改一个环境变量都值得用上下文管理器至于纯计算、无副作用的小逻辑不用也没关系。真实项目里的判断标准其实是一旦异常发生这段代码是否还能安全退出如果答案不确定那就把它包进with里。这种思路帮我减少了很多半夜被叫醒排查资源的痛苦希望你也能省去这段弯路。