Python开发者必看：这个特性让你的代码不再“泄漏”！
python约 7 分钟昨天晚上十一点多，我在公司楼下拿着奶茶吹风，手机一震，我们组那个小李给我发截图，说他那台服务的内存一路飙，最后被 K8s 直接干掉了，人还在那边说“我 Python 代码又没搞什么大对象，怎么就泄漏了？？？”
我当时困得一批，但一看他代码，笑出声：经典的资源没关干净，文件、HTTP 连接、数据库游标，全靠“良心”去 close。
说实话，这种问题我年轻那会儿也老犯。直到后面我几乎把所有“可能泄漏”的东西都换成了一个特性：with + 上下文管理器，那后面类似的故障就几乎没再遇到过了，真的是省命的。之前写数据库那篇对比的时候就顺手提过一点，今天就专门把 Python 这块拎出来聊聊。
那些年忘记关的资源：文件版“内存慢慢涨”
你们肯定写过这种代码对吧：
defread_config(path):
f = open(path, 'r', encoding='utf-8')
data = f.read()
# 这儿一堆逻辑
if"xxx"in data:
raise ValueError("配置不合法")
f.close()
return data
看着还行对不对？问题是，一旦在 raise ValueError 那里抛了异常，后面的 f.close() 是不会执行的。多来几次，这个进程里打开的文件句柄数就一直涨，最后你会遇到各种诡异错误：
打开文件突然报 Too many open files某些日志写不进去 容器内存慢慢涨，但又看不到明显大对象
然后你就开始怀疑人生。
用 with 写一遍就好看多了：
defread_config(path):
with open(path, 'r', encoding='utf-8') as f:
data = f.read()
if"xxx"in data:
raise ValueError("配置不合法")
return data
你啥都没关对吧？但是只要离开 with 那个缩进块，不管是正常 return 还是抛异常，文件一定会被关掉。资源清理这件事，从“靠自觉”变成“语言保证”，这个体验完全不一样。
with 背后到底干了啥？
很多人就会问了：with 这个东西，具体是怎么保证“不会泄漏”的？
简单说一下哈，不上特别教科书的那种讲法。
只要一个对象实现了两个方法：
classSomething:
def__enter__(self):
# 进入 with 块之前干的事
return self # 或者返回别的对象给 as 右边用
def__exit__(self, exc_type, exc_val, exc_tb):
# 离开 with 块一定会调用，不管有没有异常
# 这儿做清理：close、release、rollback...
...
Python 解释器看到：
with Something() as obj:
# 这里写你的业务逻辑
pass
内部大概就等价于这样（伪代码）：
mgr = Something()
obj = mgr.__enter__()
try:
# with 里的代码
pass
except BaseException as e:
# 有异常也会走这
mgr.__exit__(type(e), e, e.__traceback__)
raise
else:
# 没异常也会走这
mgr.__exit__(None, None, None)
所以，不管你在 with 里面是正常 return、还是抛异常、还是半路 break、continue，__exit__ 都会被调用。你只要把“释放资源”的动作塞到 __exit__ 里，就不会再因为“忘记 close”而泄漏。
你看，核心就一句话：不要再自己记得去 close，让对象自己在 __exit__ 里做“最后一件事”。
用 with 管 HTTP 连接：别让 socket 挂在那儿发呆
再说个常见的坑：HTTP 请求。
小李之前是这么写的：
import requests
deffetch_user(user_id):
resp = requests.get(f"https://api.xxx.com/users/{user_id}")
if resp.status_code != 200:
raise RuntimeError("remote error")
return resp.json()
能跑没错，但问题是 requests.get 默认会帮你做一些释放，不过如果你用的是 Session，就经常会有人忘记关：
import requests
session = requests.Session()
deffetch_user(user_id):
resp = session.get(f"https://api.xxx.com/users/{user_id}")
return resp.json()
Session 不关，连接池里那些 socket 就挂着不释放，时间一久，各种网络资源问题就出了。
更好的写法其实就是让你自己的会话变成一个上下文管理器用：
import requests
classHttpClient:
def__init__(self):
self._session = requests.Session()
def__enter__(self):
return self._session
def__exit__(self, exc_type, exc_val, exc_tb):
# 不管成功失败，记得把 Session 关掉
self._session.close()
deffetch_user(user_id):
with HttpClient() as sess:
resp = sess.get(f"https://api.xxx.com/users/{user_id}")
resp.raise_for_status()
return resp.json()
这样一来，HttpClient 的使用方是没机会“忘记 close”的，只要缩进结束，Session 必关。
数据库连接也一样：用上下文管事务，爽
再说一个特别适合用 with 的东西：数据库事务。
以前我们喜欢这么写：
deftransfer(conn, from_id, to_id, amount):
cursor = conn.cursor()
try:
cursor.execute("UPDATE account SET balance = balance - %s WHERE id = %s", (amount, from_id))
cursor.execute("UPDATE account SET balance = balance + %s WHERE id = %s", (amount, to_id))
conn.commit()
except Exception:
conn.rollback()
raise
finally:
cursor.close()
东西一多，这个 try / except / finally 就越来越长，复制粘贴到处都是。
换上上下文管理器就清爽很多：
import pymysql
classTransaction:
def__init__(self, conn):
self.conn = conn
self.cursor = None
def__enter__(self):
self.cursor = self.conn.cursor()
return self.cursor
def__exit__(self, exc_type, exc_val, exc_tb):
try:
if exc_type isNone:
self.conn.commit()
else:
self.conn.rollback()
finally:
if self.cursor isnotNone:
self.cursor.close()
deftransfer(conn, from_id, to_id, amount):
with Transaction(conn) as cursor:
cursor.execute(
"UPDATE account SET balance = balance - %s WHERE id = %s",
(amount, from_id)
)
cursor.execute(
"UPDATE account SET balance = balance + %s WHERE id = %s",
(amount, to_id)
)
你看，现在 transfer 函数里就只剩业务逻辑了，连接提交/回滚、游标关闭都被“藏”在了上下文管理器里，异常也不会导致事务悬挂不提交或者不回滚。
懒得写类？那就用 contextlib.contextmanager
有时候你就想包一小段逻辑，写一个完整的类嫌麻烦，Python 还贴心给了个装饰器：contextlib.contextmanager。
这个装饰器可以让你用生成器写上下文管理器，看一个很常用的“计时器”例子：
import time
from contextlib import contextmanager
@contextmanager
deftime_block(name: str):
start = time.time()
try:
yield
finally:
end = time.time()
print(f"[{name}] 耗时: {end - start:.3f}s")
defrun_job():
with time_block("导出报表"):
# 里面随便干事，异常也没问题
export_big_report()
这里有两个关键点：
yield之前的代码，相当于__enter__yield之后的代码，相当于__exit__，无论里面抛不抛异常，这段都会执行
你不仅可以用它做“计时”，还能做很多“用完就收尾”的事情，比如：
临时切换某个配置 临时修改 logger 的级别 临时进入一个目录再切回来
随口写一个“临时切换目录”的：
import os
from contextlib import contextmanager
@contextmanager
defcd(path: str):
old = os.getcwd()
os.chdir(path)
try:
yield
finally:
os.chdir(old)
defbuild():
with cd("/tmp/project-build"):
os.system("make all")
不管 build 的过程成功失败，最后都会回到原来的工作目录，不会出现“莫名其妙切到 /tmp 就不回来了”的问题。
多个资源一起管理：别搞 try 里套 try 那种怪物
现实里经常是多个资源一起用，比如你要：
打开一个配置文件 打开一份日志文件 建一个数据库事务
要是不用 with，你大概会变成这样：
defprocess(path, log_path, conn):
f = open(path)
try:
log = open(log_path, 'a')
try:
cursor = conn.cursor()
try:
# 一顿操作
...
finally:
cursor.close()
finally:
log.close()
finally:
f.close()
我光打这段字手都累，更别说以后谁改动的时候，少一个 finally 就出事。
用 with 可以这么写：
defprocess(path, log_path, conn):
with open(path) as f, \
open(log_path, 'a') as log, \
conn.cursor() as cursor:
# 这里写正常业务逻辑
...
同时管理三个资源，缩进还保持一级，with 会帮你按顺序调用每个对象的 __enter__ / __exit__。一旦出错，也会按照“反向顺序”一个个清理。
再进阶一点：ExitStack 应对“数量不确定”的资源
有时候，你要打开的东西数量是“运行时决定”的，比如你要把一堆文件合并：
from contextlib import ExitStack
defmerge_files(paths, out_path):
with ExitStack() as stack:
files = [stack.enter_context(open(p, 'r')) for p in paths]
with open(out_path, 'w') as out:
for f in files:
for line in f:
out.write(line)
ExitStack 的好处就是：你可以在循环里不断 enter_context 新资源，最后只用一个 with 来兜底释放。中间一旦抛异常，已经打开的那些文件也都会被安全关掉，不会出现“前 5 个关了，后 3 个还挂着”的场面。
说到底：把“清理动作”封装起来，别散落在代码里
总结这堆例子，其实都是一个套路：
找出你代码里所有需要“成对出现”的东西比如：open/close、connect/disconnect、lock/release、begin/commit、acquire/release。
把“清理那一半”统一写进一个上下文管理器类也行，
@contextmanager也行，看你喜欢。在业务代码里，只暴露一个
with XXX:让使用的人根本“没有机会忘记释放”。
以前出资源泄漏，排查都是这样：先怀疑 Python GC、再怀疑第三方库，最后一圈兜兜转转发现——是自己忘了 close。 把这一类东西系统地改成 with 之后，日志干净了，内存图也平滑了，人也睡得更香了。
行了，我这会儿咖啡也凉了，差不多先聊到这。
你可以先翻一翻你现在在维护的项目，搜一下 open(、Session(、cursor(、connect( 这些关键词，看有没有一大片 close() 散落在各个函数里的，那些地方，基本都能用上下文管理器收一收。
有啥具体场景搞不定，你扔一段代码过来，我帮你一起改成不“泄漏”的版本，顺便再吐槽两句。