文件操作是每个程序都逃不掉的事情。但 "打开文件 → 读写 → 关闭" 这个流程里，藏着很多坑：忘记关闭导致文件句柄泄漏、异常发生时资源没清理、代码嵌套层级越来越深。with 语句就是 Python 为解决这些问题提供的语法糖。
本章从文件读写讲起，逐步深入到 with 的底层协议、自定义上下文管理器，以及 contextlib 工具链，让你写出既安全又优雅的资源管理代码。
指自己使用 open() 打开文件，使用 close() 关闭文件，亲力亲为。例如：
f = open("data.txt", "r", encoding="utf-8")
data = f.read()
f.close()
print(data)上面的代码有三个问题：
f.read() 如果抛出异常，f.close() 永远不会执行，文件句柄泄漏；
如果代码逻辑复杂，close() 可能被遗忘；
需要显式处理编码，否则默认编码因平台而异（Windows 可能是 GBK，Linux 是 UTF-8）。
为了解决抛出异常，文件句柄泄露，我们可以这样做：
f = None
try:
f = open("data.txt", "r", encoding="utf-8")
data = f.read()
finally:
# 即使报错也会被执行
if f is not None:
f.close()
print(data)上面代码，在抛出错误是也能正确关闭文件句柄。但是，每次操作文件都要写 try-finally 不麻烦吗？你可以选择将代码封装在一个函数中，减少代码的编写，像这样：
def read_file(file_path, encoding="utf-8"):
f = None
try:
f = open(file_path, "r", encoding=encoding)
data = f.read()
return data
finally:
if f is not None:
f.close()
# 调用
content = read_file("data.txt")
print(content)虽然解决了重复编写，但是代码量变多了。其实不用这么麻烦，Python 早就想到了，提供了一个 with 语法。
with 是 Python 提供的资源安全管理语法糖，替代手写 try-finally。
进入代码块自动执行准备逻辑，离开代码块（正常结束 / 抛出异常 /return）自动执行清理逻辑。
文件场景最典型：自动调用 .close()，杜绝忘记关闭文件、连接泄漏。
例如：
with open("data.txt", "r", encoding="utf-8") as f:
text = f.read()
print(text)
# 出了 with 块，f 已经自动关闭，不需要手动调用 f.close()with 的核心保证是：无论 with 块内的代码是正常结束还是抛出异常，文件对象都会被关闭。这等价于下面的 try/finally：
f = open("data.txt", "r", encoding="utf-8")
try:
text = f.read()
finally:
f.close()with 让这段样板代码浓缩成一行，而且异常安全。
下面是对文件操作的模式速记表，前面基础章节已经介绍过：
| 模式 | 含义 | 文件存在 | 文件不存在 | 指针位置 |
| r | 只读（默认） | 打开 | 报错 | 开头 |
| w | 只写 | 截断清空 | 创建 | 开头 |
| x | 独占创建 | 报错 | 创建 | 开头 |
| a | 追加写 | 打开 | 创建 | 末尾 |
| b | 二进制模式 | 与上述组合 | 与上述组合 | — |
| + | 读写模式 | 与上述组合 | 与上述组合 | — |
常见组合：
rb：读取二进制文件（图片、视频、压缩包）；
w+：读写，先清空再写；
a+：读写，追加模式，指针在末尾；
r+：读写，不清空，指针在开头，可覆盖已有内容。
对于大文件，如 10GB 的大文件，不要使用 read() 进行读取，可能一下就让电脑崩溃。正确的姿势应该是逐行读取，这样内存消耗最小。
例如：下面提供三种按行读取的方式，其中方式二对大文件不可取。
# 方式一：迭代文件对象（推荐，内存占用最小）
with open("data.txt", "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
print(line)
# 方式二：readlines()（加载全部到内存，不适合大文件）
with open("data.txt", "r", encoding="utf-8") as f:
lines = f.readlines() # 返回列表
print(lines)
# 方式三：readline()（手动逐行，更可控）
with open("data.txt", "r", encoding="utf-8") as f:
while True:
line = f.readline()
if not line:
break
print(line.strip())注意：方式一是 Python 中读取大文件的最佳实践。文件对象本身就是迭代器，每次 yield 一行，不会把整个文件载入内存。配合 with 语句，文件在循环结束后自动关闭。
以写入（w）模式打开文件时，如果文件已存在，立刻清空文件内所有原有内容。如果文件不存在，则创建新文件，随后从头开始写入数据。适合需要覆盖旧内容、生成全新文件的场景。注意，一旦以 w 打开，还没写入任何内容，旧数据就已经被清空。
以追加（a）模式打开文件时不会清空原有内容，如果文件不存在，则创建。文件打开后，文件指针固定在文件末尾，所有新内容都接续在原有文字后面。适合日志记录、持续新增数据，保留历史内容。
例如：
# 写入（覆盖已有内容）
with open("output.txt", "w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
# 追加（不覆盖，追加到末尾）
with open("output.txt", "a", encoding="utf-8") as f:
f.write("第三行\n")
# 写入多行
lines = ["apple\n", "banana\n", "cherry\n"]
with open("output.txt", "w", encoding="utf-8") as f:
f.writelines(lines) # 注意：列表元素需要自带换行符
# 更推荐的方式：print 的 file 参数
with open("output.txt", "w", encoding="utf-8") as f:
for item in ["apple", "banana", "cherry"]:
print(item, file=f) # print 自动添加换行二进制模式下，read() 返回 bytes，write() 接收 bytes。文本模式下，Python 自动做编码解码。处理网络数据、图片、音频时，二进制模式是必须的。
例如：
# 读取二进制文件
with open("image.png", "rb") as f:
header = f.read(8) # 读取前 8 字节（PNG 文件头）
print(header) # b'\x89PNG\r\n\x1a\n'
# 写入二进制文件
with open("data.bin", "wb") as f:
f.write(b"\x00\x01\x02\x03")
f.write(bytes([65, 66, 67]))
# 文本模式 vs 二进制模式
text = "你好"
with open("utf8.txt", "w", encoding="utf-8") as f:
f.write(text)
with open("utf8.txt", "rb") as f:
raw = f.read()
print(raw) # b'\xe4\xbd\xa0\xe5\xa5\xbd' —— 6 字节 UTF-8 编码with 语句能自动关闭文件，不是 Python 对文件的特殊照顾，而是靠一个通用的协议：上下文管理器协议（Context Manager Protocol）。任何实现了 __enter__ 和 __exit__ 两个魔法方法的对象，都可以用在 with 语句中。
上下文管理协议提供了两个函数，__enter__(self) 进入 with 块时调用，返回值赋给 as 变量，__exit__(self, exc_type, exc_val, exc_tb) 退出 with 块时调用，无论是否发生异常。例如：
class ManagedResource:
def __init__(self, name: str):
self.name = name
self.active = False
def __enter__(self):
"""进入 with 块时调用，返回值赋给 as 变量"""
print(f"[{self.name}] 资源初始化")
self.active = True
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""退出 with 块时调用，无论是否发生异常"""
print(f"[{self.name}] 资源清理")
self.active = False
# 返回 False 让异常继续传播，返回 True 则吞掉异常
return False
def do_work(self):
print(f"[{self.name}] 工作中...")
# 使用
with ManagedResource("数据库连接") as res:
res.do_work()
print(f"with 块内 active={res.active}")
# 输出：资源清理，active=False
# 等价的手动写法（但 with 更简洁）
res = ManagedResource("手动模式")
res_obj = res.__enter__()
try:
res_obj.do_work()
finally:
res.__exit__(None, None, None)运行示例，输出：
[数据库连接] 资源初始化
[数据库连接] 工作中...
with 块内 active=True
[数据库连接] 资源清理
[手动模式] 资源初始化
[手动模式] 工作中...
[手动模式] 资源清理注意，__exit__ 接收三个参数：
exc_type：异常类型，没有异常时为 None；
exc_val：异常对象；
exc_tb：traceback 对象。
如果 __exit__ 返回 True，with 块内的异常被压制，不会向上传播。这在某些容错场景有用，但通常应该返回 False，让异常暴露出来。如果不让异常暴露出来，对排查错误不友好，排查半天都不知道具体错误是什么。
下面示例演示，with 语句块中排出了异常信息，但是资源依然被正确释放，例如：
class SafeWriter:
def __init__(self, path: str):
self.path = path
self.fp = None
def __enter__(self):
"""打开文件"""
self.fp = open(self.path, "w", encoding="utf-8")
print(f"[SafeWriter] 打开 {self.path}")
return self.fp
def __exit__(self, exc_type, exc_val, exc_tb):
"""释放资源"""
if self.fp:
self.fp.close()
print(f"[SafeWriter] 关闭 {self.path}")
if exc_type:
print(f"[SafeWriter] 捕获异常: {exc_type.__name__}: {exc_val}")
return False # 不吞异常
# 使用
with SafeWriter("test.txt") as f:
f.write("hello\n")
raise ValueError("模拟写入出错") # 即使抛异常，文件仍被关闭运行代码，输出：
[SafeWriter] 打开 test.txt
[SafeWriter] 关闭 test.txt
[SafeWriter] 捕获异常: ValueError: 模拟写入出错
Traceback (most recent call last):
File "d:\share_dir\workspace\5.demo\python_demo\demo.py", line 24, in <module>
raise ValueError("模拟写入出错") # 即使抛异常，文件仍被关闭
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ValueError: 模拟写入出错看到了吗？这就是 with 的核心价值：异常发生时，资源清理仍然执行。对于锁、数据库连接、文件句柄等稀缺资源，这个保证至关重要。
掌握了协议之后，你可以为任何需要 "先初始化、后清理" 的场景写上下文管理器。下面分别演示锁、数据库、网络管理器的封装，可以作为参考。
通过封装上下文管理器，避免手动重复调用 acquire() 加锁，release() 释放锁。例如：
# context_lock.py
import threading
import time
class NamedLock:
def __init__(self, name: str):
self.name = name
self._lock = threading.Lock()
def __enter__(self):
self._lock.acquire()
print(f"[Lock] {self.name} 已获取")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self._lock.release()
print(f"[Lock] {self.name} 已释放")
return False
# 使用
def worker(name: str):
with NamedLock("共享资源"):
print(f"[{name}] 访问共享资源")
time.sleep(0.5)
threads = [threading.Thread(target=worker, args=(f"Thread-{i}",)) for i in range(3)]
for t in threads: t.start()
for t in threads: t.join()运行代码，输出：
[Lock] 共享资源 已获取
[Thread-0] 访问共享资源
[Lock] 共享资源 已获取
[Thread-1] 访问共享资源
[Lock] 共享资源 已获取
[Thread-2] 访问共享资源
[Lock] 共享资源 已释放[Lock] 共享资源 已释放
[Lock] 共享资源 已释放实际上，Python 的 threading.Lock 本身已经实现了上下文管理器协议，可以直接 with lock:。上面的例子只是为了展示协议如何工作。真实代码中，直接写 with threading.Lock(): 即可。
和上面锁同样的道理，也是用上下文管理器实现数据库连接和释放连接，以及事务回滚和提交操作。例如：
import sqlite3
from typing import Optional
class Database:
def __init__(self, db_path: str):
self.db_path = db_path
self.conn: Optional[sqlite3.Connection] = None
def __enter__(self) -> sqlite3.Connection:
"""建立连接"""
self.conn = sqlite3.connect(self.db_path)
print(f"[DB] 连接 {self.db_path}")
return self.conn
def __exit__(self, exc_type, exc_val, exc_tb):
"""释放连接，或提交/回滚事务"""
if exc_type:
self.conn.rollback()
print(f"[DB] 事务回滚")
else:
self.conn.commit()
print(f"[DB] 事务提交")
self.conn.close()
print(f"[DB] 连接关闭")
return False
# 使用
with Database(":memory:") as conn:
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
conn.execute("INSERT INTO users (name) VALUES ('Alice')")
# 正常退出时自动提交并关闭
# 异常时自动回滚
with Database(":memory:") as conn:
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
conn.execute("INSERT INTO users (name) VALUES ('Bob')")
raise RuntimeError("模拟数据库错误")
# 输出：事务回滚 → 连接关闭运行代码，输出：
[DB] 连接 :memory:
[DB] 事务提交
[DB] 连接关闭
[DB] 连接 :memory:
[DB] 事务回滚
[DB] 连接关闭
Traceback (most recent call last):
File "d:\share_dir\workspace\5.demo\python_demo\demo.py", line 37, in <module>
raise RuntimeError("模拟数据库错误")
RuntimeError: 模拟数据库错误这个设计比手动 try/conn.commit()/conn.close() 优雅得多。__exit__ 根据是否有异常来决定提交还是回滚，是数据库事务的典型模式。
同理，使用上下文管理器实现网络连接的建立和断开。例如：
import socket
import time
from typing import Optional
class TcpClient:
def __init__(self, host: str, port: int, timeout: float = 5.0):
self.host = host
self.port = port
self.timeout = timeout
self.sock: Optional[socket.socket] = None
def __enter__(self) -> socket.socket:
"""建立网络连接"""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(self.timeout)
self.sock.connect((self.host, self.port))
print(f"[TCP] 已连接 {self.host}:{self.port}")
return self.sock
def __exit__(self, exc_type, exc_val, exc_tb):
"""断开网络连接"""
if self.sock:
self.sock.close()
print(f"[TCP] 已断开 {self.host}:{self.port}")
if exc_type:
print(f"[TCP] 连接异常: {exc_type.__name__}")
return False
# 使用（假设有一个 Echo 服务器在 127.0.0.1:65432）
try:
with TcpClient("127.0.0.1", 65432, timeout=2.0) as sock:
sock.sendall(b"hello")
data = sock.recv(1024)
print(f"收到: {data.decode()}")
except (ConnectionRefusedError, socket.timeout) as e:
print(f"连接失败: {e}")运行代码，输出：
[TCP] 已连接 127.0.0.1:65432
收到: hello
[TCP] 已断开 127.0.0.1:65432TcpClient 封装了连接建立和断开，无论通信成功还是失败，socket 都会被关闭。这比手动 try/sock.close() 少写很多样板代码，而且更难出错。
注意了，这里启动的 Echo 服务是一个使用 Python 实现的简单服务，点击获取完整代码。
# echo_server.py
import socket
# 1. 创建TCP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 允许端口快速复用，避免重启报错
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 2. 绑定地址与端口
HOST = "127.0.0.1"
PORT = 65432
server_socket.bind((HOST, PORT))
# 3. 开始监听
server_socket.listen()
print(f"Echo 服务器启动 {HOST}:{PORT}，等待客户端连接...")
# 4. 循环接受连接
while True:
conn, addr = server_socket.accept()
print(f"客户端接入：{addr}")
with conn:
while True:
# 接收数据，缓冲区1024字节
data = conn.recv(1024)
if not data:
# 收到空 = 客户端关闭连接
print(f"客户端 {addr} 断开连接")
break
print("收到:", data.decode("utf-8"))
# echo：原样发回
conn.sendall(data)如果你觉得手写 __enter__/__exit__ 比较啰嗦。可以使用 contextlib 模块提供了 @contextmanager 装饰器，让你用生成器函数写上下文管理器。
from contextlib import contextmanager
@contextmanager
def managed_file(path: str, mode: str = "r"):
"""等价于上面的 SafeWriter，但代码量只有一半"""
f = open(path, mode, encoding="utf-8")
print(f"[CM] 打开 {path}")
try:
yield f # yield 的值就是 as 变量
finally:
f.close()
print(f"[CM] 关闭 {path}")
# 使用
with managed_file("output.txt", "w") as f:
f.write("contextmanager 写入\n")
# 出了 with 块，自动执行 finally运行代码，输出：
[CM] 打开 output.txt
[CM] 关闭 output.txt@contextmanager 的工作原理：
函数执行到 yield 之前，相当于 __enter__。即相当于如下代码：
f = open(path, mode, encoding="utf-8")
print(f"[CM] 打开 {path}")被放在 __enter__ 中：
def __enter__(self):
f = open(path, mode, encoding="utf-8")
print(f"[CM] 打开 {path}")
return fyield 的值赋给 as 变量，相当于 __enter__ 中 return 语句的作用。
with 块结束后，执行 yield 之后的代码，相当于 __exit__。即相当于如下代码：
f.close()
print(f"[CM] 关闭 {path}")被放在 __exit__ 中执行：
def __exit__(self, exc_type, exc_val, exc_tb):
f.close()
print(f"[CM] 关闭 {path}")如果 with 块抛异常，异常会在 yield 处被抛出，可以用 try/except/finally 捕获。
和自定义上下文管理器协议实现一样，@contextmanager 定义的上下文管理器也支持传递参数，例如：
from contextlib import contextmanager
import time
@contextmanager
def timed_block(name: str = "Block"):
start = time.time()
print(f"[{name}] 开始...")
try:
yield # 不需要返回值，as 变量为 None
finally:
elapsed = time.time() - start
print(f"[{name}] 完成，耗时: {elapsed:.4f}s")
# 使用
with timed_block("数据导入"):
time.sleep(0.3)
print("导入中...")
with timed_block("数据清洗"):
time.sleep(0.2)运行代码，输出：
[数据导入] 开始...
导入中...
[数据导入] 完成，耗时: 0.3009s
[数据清洗] 开始...
[数据清洗] 完成，耗时: 0.2002scontextlib.suppress 是 Python 3.4+ 的快捷工具，用于有预期地忽略异常。注意：它只压制指定的异常类型，其他异常仍然传播。例如：
from contextlib import suppress
# 忽略特定异常，代码更简洁
with suppress(FileNotFoundError):
with open("不存在的文件.txt", "r") as f:
print(f.read())
# 如果文件不存在，不会抛异常，静默跳过
# 等价于：
try:
with open("不存在的文件.txt", "r") as f:
print(f.read())
except FileNotFoundError:
pass当你需要同时管理多个资源时，ExitStack 能自动按后进先出的顺序清理。例如：
from contextlib import ExitStack
with ExitStack() as stack:
f1 = stack.enter_context(open("a.txt", "w", encoding="utf-8"))
f2 = stack.enter_context(open("b.txt", "w", encoding="utf-8"))
f3 = stack.enter_context(open("c.txt", "w", encoding="utf-8"))
f1.write("文件 A\n")
f2.write("文件 B\n")
f3.write("文件 C\n")
# 出了 with 块，f3、f2、f1 依次自动关闭
# 动态数量的资源管理
files = ["log1.txt", "log2.txt", "log3.txt"]
with ExitStack() as stack:
fps = [stack.enter_context(open(name, "w", encoding="utf-8")) for name in files]
for fp in fps:
fp.write("日志条目\n")
# 所有文件自动关闭ExitStack 特别适合资源数量在运行时才确定的情况（比如批量打开多个文件）。它保证所有 enter_context 注册的资源都会在退出时关闭，即使有资源在打开过程中失败，已经打开的资源也会被正确清理。
文件内部有一个文件指针（也叫读写光标），后续 read() / write() 都从指针当前位置开始操作。
tell() 查询指针当前位置，返回字节偏移量。
seek(offset, whence) 移动文件指针，实现随机读写。
offset：偏移量。
whence：起始位置（0=文件开头，1=当前位置，2=文件末尾）。
重要提醒：
在文本模式下（默认），seek 只能跳到 tell() 返回的位置，不能随意跳到字节偏移，因为文本编码下字符和字节不一定一一对应，部分平台不能随意向前 seek。
随机访问标准做法：以二进制模式 rb / wb / ab+ 打开文件。
with open("data.bin", "wb") as f:
f.write(b"ABCDEFGHIJ")
with open("data.bin", "rb") as f:
print(f"位置: {f.tell()}") # 0
f.seek(5) # 跳到第 5 字节（从 0 开始）
print(f.read(3)) # b'FGH'
print(f"位置: {f.tell()}") # 8
f.seek(-3, 2) # 从末尾往回 3 字节
print(f.read()) # b'H'运行代码，返回：
位置: 0
b'FGH'
位置: 8
b'HIJ'截断文件，即修改文件占用磁盘长度。
如果 size < 当前文件大小：尾部数据直接丢弃，文件变短；
如果 size > 当前文件大小：文件扩容，中间填充空字节 \x00；
如果 truncate() 不传参数，则以当前文件指针位置作为新长度。
注意：truncate() 生效依赖可写打开模式 r+ / w+ / a+。
例如：
with open("output.txt", "w+") as f:
f.write("1234567890")
f.seek(5)
f.truncate() # 截断到当前位置，文件变为 "12345"运行后，output.txt 文件内容如下：
普通文件读写，使用 read() 将磁盘数据拷贝到进程内存缓冲区。对于读写性能要求非常高的场景就会成为瓶颈。
此时，可以使用内存映射 mmap，操作系统把磁盘文件直接映射到进程虚拟地址空间。程序操作内存地址，等价于直接操作文件，操作系统自动负责磁盘与内存同步。这样可以：
减少一次数据拷贝，大文件读写性能更高；
支持像数组一样随机访问 mmap_obj[i]；
支持切片、seek、切片修改；
多个进程映射同一个文件可实现进程间通信。
例如：
import mmap
with open("data.bin", "r+b") as f:
with mmap.mmap(f.fileno(), 0) as mm:
print(mm[:10]) # 像访问字节串一样访问文件，不复制到内存运行代码，输出：
b'ABCDEFGHIJ'临时文件与目录指程序运行期间临时存放数据，用完自动销毁。
标准库 tempfile，会自动处理路径、权限、清理，不要自己手动在 /tmp 或桌面创建文件。使用 tempfile 库，会：
自动选择系统临时目录（Windows：%TEMP%，Linux/macOS：/tmp）
生成唯一文件名，避免重名冲突
支持自动删除（进程退出 / 上下文结束自动清理）
例如：
import tempfile
import os
# 临时文件，自动删除
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=True) as f:
f.write("临时数据\n")
print(f"临时文件路径: {f.name}")
# 出了 with 块，文件自动删除
# 临时目录
with tempfile.TemporaryDirectory() as tmpdir:
path = os.path.join(tmpdir, "report.txt")
with open(path, "w") as f:
f.write("报告内容")
print(f"目录内文件: {os.listdir(tmpdir)}")
# 出了 with 块，整个目录被删除
# 手动控制删除（delete=False）
with tempfile.NamedTemporaryFile(mode="w", delete=False) as f:
f.write("持久化临时数据\n")
persistent_path = f.name
print(f"文件保留在: {persistent_path}")
os.unlink(persistent_path) # 手动删除运行代码，输出：
临时文件路径: C:\Users\ADMINI~1\AppData\Local\Temp\tmprhwzw7vw.txt
目录内文件: ['report.txt']
文件保留在: C:\Users\ADMINI~1\AppData\Local\Temp\tmp9440ipuc注意，tempfile 模块是处理临时数据的标准做法，比手动在 /tmp 下创建文件更安全（自动避免文件名冲突，权限控制更严格）。
川公网安备51010802032098