Python 基础体系 · 第 38/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python I/O 模型：文本、二进制、缓冲、编码和流式处理

Python 中的 I/O，表面上是“打开文件、读写内容”，底层却至少涉及五个不同问题：

1. 数据到底是 str 还是 bytes；
2. 字节如何解释为字符，也就是编码和解码；
3. 数据是否经过缓冲，以及何时真正到达操作系统；
4. 换行符是否被转换；
5. 数据是一次性加载，还是以流的形式分段处理。

如果把这些概念混在一起，就容易出现典型错误：用文本模式读取图片、把字符数当成字节数、以为 write() 已经把数据写入磁盘、按固定字节数切分 UTF-8 文本，或者在网络流上错误地假设一次 recv() 就能得到一条完整消息。

本文以 Python 3.14 的 io 模块和内置 open() 为基础，建立一套从操作系统文件描述符到 Python 字符串的完整模型。

---

一、先建立数据流的分层模型

Python 的文件对象不是单一类型，而是多个层次组合出来的对象。最常见的文本文件对象可以抽象为：

Python str
   │
   │  编码/解码、错误处理、换行转换
   ▼
TextIOWrapper
   │
   │  二进制缓冲
   ▼
BufferedReader / BufferedWriter / BufferedRandom
   │
   │  系统调用，文件描述符
   ▼
FileIO
   │
   ▼
操作系统文件、管道、终端、设备或其他 I/O 端点

io 模块把流分为文本 I/O、二进制 I/O 和原始 I/O。文本流处理 str，二进制流处理 bytes 或其他 bytes-like 对象，原始流则提供更接近操作系统的低级字节读写接口。TextIOWrapper 通常包装一个缓冲二进制流，而缓冲二进制流又包装一个原始流。(docs.python.org)

可以用下面的代码观察普通文件对象的层次：

from pathlib import Path

path = Path("example.txt")
path.write_text("你好\n", encoding="utf-8")

with path.open("r", encoding="utf-8") as text_stream:
    print(type(text_stream))
    print(type(text_stream.buffer))
    print(type(text_stream.buffer.raw))

一种常见输出是：

<class '_io.TextIOWrapper'>
<class '_io.BufferedReader'>
<class '_io.FileIO'>

具体的内部类名属于实现细节，但“文本层—缓冲二进制层—原始文件层”是理解普通文件 I/O 的关键结构。

1. 流不一定是文件

“文件对象”只是历史名称。流可以来自：

• 磁盘文件；
• 标准输入、标准输出和标准错误；
• 管道；
• socket；
• 内存中的 BytesIO 或 StringIO；
• 设备文件；
• 第三方库提供的 file-like object。

不同流的能力并不相同：

• 只读、只写或读写；
• 支持随机定位，或者只能顺序访问；
• 阻塞或非阻塞；
• 是否支持 fileno()；
• 是否能可靠地判断总长度。

因此，能够调用 read() 不代表一定能够调用 seek()；能够写入数据，也不代表数据已经持久化到存储设备。

---

二、文本、二进制和原始 I/O 的边界

1. 文本 I/O 处理 str

文本流向调用者提供字符接口：

with open("message.txt", "w", encoding="utf-8") as f:
    f.write("你好，Python\n")

这里的 f.write() 接受 str，返回写入的字符数：

with open("message.txt", "w", encoding="utf-8") as f:
    count = f.write("你好")

print(count)

输出：

2

这不是写入的字节数。字符串 "你好" 使用 UTF-8 编码后占 6 个字节：

data = "你好".encode("utf-8")

print(len("你好"))       # 2
print(len(data))         # 6
print(data)              # b'\xe4\xbd\xa0\xe5\xa5\xbd'

str 的长度表示 Unicode 字符串中包含的代码点数量，而编码后的 bytes 长度取决于编码方式。工程代码不能把两者混为一谈。

2. 二进制 I/O 处理 bytes

二进制流不负责编码、解码或换行转换：

with open("data.bin", "wb") as f:
    f.write(b"\x00\x01\xff")

读取时得到的是 bytes：

with open("data.bin", "rb") as f:
    data = f.read()

print(type(data))  # <class 'bytes'>
print(data)       # b'\x00\x01\xff'

如果向二进制流传入字符串，Python 会直接拒绝：

with open("data.bin", "wb") as f:
    f.write("文本")

结果：

TypeError: a bytes-like object is required, not 'str'

这个错误不是 Python 缺少自动转换，而是有意阻止隐式编码。因为“应该使用什么编码”无法从任意字符串和文件对象中可靠推断出来。

反过来，文本流也不接受 bytes：

with open("message.txt", "w", encoding="utf-8") as f:
    f.write(b"hello")

结果同样是 TypeError。

3. 原始 I/O 不等于“没有任何缓存的普通文件操作”

原始 I/O 由 RawIOBase 表示，普通文件对应的具体类型通常是 FileIO。可以通过 buffering=0 创建原始二进制流：

with open("data.bin", "rb", buffering=0) as f:
    print(type(f))

输出通常是：

<class '_io.FileIO'>

原始流的特点是更接近操作系统调用：

• read(size) 可能少于 size；
• write(data) 可能只写入一部分；
• 调用方需要检查返回值并决定是否重试。

例如，原始写入接口并不保证一次写完全部数据：

def write_all(raw_stream, data):
    view = memoryview(data)
    while view:
        n = raw_stream.write(view)
        if n is None:
            continue
        if n == 0:
            raise OSError("stream made no progress")
        view = view[n:]

对磁盘文件来说，普通情况下短写不一定经常出现；但对于管道、设备、非阻塞流或自定义流，不能据此省略处理。缓冲二进制层和文本层通常会负责更高层的重试逻辑。(docs.python.org)

---

三、编码和解码：文本边界在哪里

1. str 与 bytes 是两个不同的数据域

可以把编码过程写成：

str --encode(encoding)--> bytes
bytes --decode(encoding)--> str

例如：

text = "A中"

encoded = text.encode("utf-8")
decoded = encoded.decode("utf-8")

print(encoded)          # b'A\xe4\xb8\xad'
print(decoded == text)  # True

在编码一致且数据没有损坏时：

decode(encode(text, E), E) = text

其中 E 是同一个字符编码。

但不同编码产生的字节序列不同：

text = "中"

print(text.encode("utf-8"))
print(text.encode("utf-16"))

因此，bytes 本身通常不携带“这是 UTF-8”或“这是 GBK”的元数据。解码方必须知道或推断编码。

2. 错误处理策略会改变数据语义

文本 I/O 的 errors 参数决定编码或解码失败时怎么办。常见策略包括：

• strict：遇到错误立即抛出异常；
• ignore：丢弃无法处理的数据；
• replace：使用替换字符；
• backslashreplace：转成反斜杠转义形式。

示例：

bad = b"hello\xffworld"

for errors in ("strict", "ignore", "replace", "backslashreplace"):
    try:
        result = bad.decode("utf-8", errors=errors)
        print(errors, repr(result))
    except UnicodeDecodeError as exc:
        print(errors, type(exc).__name__)

可能得到：

strict UnicodeDecodeError
ignore 'helloworld'
replace 'hello�world'
backslashreplace 'hello\\xffworld'

ignore 看似让程序继续运行，实际上已经造成不可逆的信息丢失。日志展示、容错导入和数据迁移可以有不同策略，但核心数据处理通常应使用 strict，让损坏尽早暴露。

3. 默认编码不是稳定的跨平台协议

open() 和 TextIOWrapper 在没有显式指定编码时，默认使用与环境相关的编码；Python 3.14 文档明确建议，对明确使用 UTF-8 的文本文件传入 encoding="utf-8"。encoding="locale" 可以显式表示使用当前 locale 编码。(docs.python.org)

不推荐：

with open("config.json") as f:
    config = f.read()

推荐：

with open("config.json", encoding="utf-8") as f:
    config = f.read()

如果正在设计一个库函数，不希望直接替调用者决定编码，可以使用 io.text_encoding() 转发默认行为，并在启用 EncodingWarning 时帮助调用者发现未指定编码的问题。不过对于新接口，通常应明确约定 UTF-8，而不是把系统 locale 当成协议。(docs.python.org)

可以用下面的命令检查潜在的默认编码使用：

python -X warn_default_encoding script.py

也可以通过环境变量启用相应警告：

PYTHONWARNDEFAULTENCODING=1 python script.py

警告只能帮助诊断，不能替代编码协议。对于 JSON、CSV、Markdown、配置文件和跨机器交换数据，应在接口或文件格式层面明确编码。

---

四、换行符是文本层的另一个转换

换行处理与编码不同，但同样发生在文本层。

TextIOWrapper 的 newline 参数控制读写时的换行行为：

newline	读取	写入
None	启用通用换行，\r、\n、\r\n 都转换为 \n	\n 转为系统默认换行符
""	启用通用换行，但保留原始换行符	不转换
"\n"	只按 \n 识别，保留换行符	不转换
"\r"	只按 \r 识别，保留换行符	\n 转为 \r
"\r\n"	只按 \r\n 识别，保留换行符	\n 转为 \r\n

这些规则由文本流的 newline 参数定义。(docs.python.org)

1. 默认文本读取会规范化换行

假设文件的原始字节内容是：

a\r\nb\rc\n

使用默认的文本读取：

from pathlib import Path

Path("lines.txt").write_bytes(b"a\r\nb\rc\n")

with open("lines.txt", encoding="ascii", newline=None) as f:
    print(repr(f.read()))

结果是：

'a\nb\nc\n'

如果需要判断原始文件使用了哪些换行符：

with open("lines.txt", encoding="ascii", newline="") as f:
    text = f.read()
    print(repr(text))
    print(repr(f.newlines))

这里 newline="" 使换行符保留在返回字符串中；f.newlines 反映已经观察到的换行类型。

2. 二进制模式不会替换换行

with open("lines.txt", "rb") as f:
    data = f.read()

print(repr(data))

结果仍然是：

b'a\r\nb\rc\n'

所以需要保留文件原始字节时，应使用二进制模式，而不是依赖文本模式中的特殊配置。

3. Path.read_text() 也支持换行参数

Python 3.13 为 Path.read_text() 增加了 newline 参数；Path.open() 的相关参数与内置 open() 一致。(docs.python.org)

from pathlib import Path

text = Path("lines.txt").read_text(
    encoding="utf-8",
    newline=""
)

Path.read_text() 会打开文件、读取文本并关闭文件；如果需要逐块读取、逐行处理或多次操作，应使用 Path.open() 配合 with，而不是一次性便利方法。

---

五、缓冲：减少系统调用，不等于持久化

1. 缓冲的目的

系统调用和设备访问通常比 Python 内存中的复制、切片或函数调用更昂贵。缓冲层把多次小读写合并成较少的底层操作：

多次小写入
    ▼
Python 缓冲区
    ▼ flush()
较少次数的系统调用

缓冲提高了吞吐，但引入了一个状态：Python 代码已经调用了 write()，数据可能仍停留在用户态缓冲区。

f = open("output.txt", "w", encoding="utf-8")
f.write("hello")
# 此时不能简单推出数据已经到达磁盘
f.flush()
f.close()

flush() 通常把 Python I/O 层的缓冲数据推向底层对象；close() 会先执行必要的刷新，再关闭流。至于操作系统页缓存何时写入物理介质，则是更低层的持久化问题，不能把普通 flush() 等同于断电安全。

2. with 管理的是生命周期

正确的文件生命周期通常写成：

with open("output.txt", "w", encoding="utf-8") as f:
    f.write("hello\n")

离开 with 块时，文件对象会被关闭，即使代码在块内抛出异常也会执行清理。

不推荐依赖析构函数：

f = open("output.txt", "w", encoding="utf-8")
f.write("hello")
# 依赖变量离开作用域或垃圾回收来关闭

文件描述符是有限资源；在长时间运行的服务中，延迟关闭可能造成资源耗尽，而异常路径还可能使数据没有按预期刷新。

3. 三种容易混淆的“立即”

“立即写入”至少有三个不同含义：

1. TextIOWrapper.write() 立即返回；
2. 文本层数据立即交给底层二进制缓冲；
3. 数据已经由操作系统写入稳定存储。

TextIOWrapper 的 write_through=True 表示文本层写入不在 TextIOWrapper 自身缓冲，但底层二进制缓冲仍可能存在；line_buffering=True 则在包含换行或回车的写入后隐含调用 flush()。这些选项不能直接承诺物理介质持久化。(docs.python.org)

对于交互式终端，行缓冲通常改善用户看到输出的及时性：

import sys

sys.stdout.reconfigure(line_buffering=True)
print("processing...")

对于普通磁盘文件，频繁 flush() 可能降低性能；是否刷新应由协议、可见性要求和故障模型决定，而不是把 flush() 当成通用安全开关。

---

六、open() 的模式、对象和状态

常见模式可以拆成三个维度：

• 基础操作：r 读取、w 覆盖写入、a 追加、x 独占创建；
• 数据类型：默认文本，加入 b 后为二进制；
• 读写权限：加入 + 后允许读写。

例如：

open("a.txt", "r", encoding="utf-8")   # 文本读取
open("a.txt", "rb")                    # 二进制读取
open("a.txt", "w", encoding="utf-8")   # 文本覆盖写
open("a.txt", "ab")                    # 二进制追加
open("a.txt", "x", encoding="utf-8")   # 文件必须不存在
open("a.txt", "r+", encoding="utf-8")  # 文本读写，不自动截断

w 会截断已有文件；如果需要“文件不存在才创建”，应使用 x，或者采用临时文件加替换的写入流程。

1. 读写流有当前位置

对支持随机访问的文件，读写操作会改变当前位置：

from pathlib import Path

Path("sample.bin").write_bytes(b"abcdef")

with open("sample.bin", "rb") as f:
    print(f.read(2))  # b'ab'
    print(f.tell())   # 2
    print(f.seek(4))  # 4
    print(f.read(1))  # b'e'

这里 seek(4) 的偏移量以字节为单位，因为这是二进制流。

文本流更复杂：

with open("sample.txt", "r", encoding="utf-8") as f:
    position = f.tell()
    part = f.read(3)
    f.seek(position)

文本流的 tell() 返回的是不透明位置标记，通常不能当作底层字节偏移量；TextIOBase.seek() 只对特定形式的偏移有效，尤其不能随意用字符数代替字节位置。(docs.python.org)

2. 文本读写模式下的随机定位限制

UTF-8 是变长编码，一个 Unicode 字符可能对应 1 到 4 个字节。假设：

text = "A中B"
data = text.encode("utf-8")

字节布局是：

A       中                 B
41      e4 b8 ad           42

如果把“第 2 个字符”错误地当作“第 2 个字节”，就会落在多字节字符的中间。此时无法直接从任意字节位置解码出合法文本。

因此，文本流的随机访问必须遵守其不透明位置协议；需要精确按字节定位时，应使用二进制流，并在确定边界后自行解码。

---

七、流式处理：不要把“读完”误认为“处理完”

1. 一次性读取的内存模型

with open("large.log", encoding="utf-8") as f:
    content = f.read()

这种方式的峰值内存至少受到完整文本内容影响，还可能因为后续切分、复制、解析产生额外对象。

Path.read_text() 和 Path.read_bytes() 都是一次性读取便利方法：前者返回解码后的字符串，后者返回完整 bytes。(docs.python.org)

对于小型配置文件，这种写法简单可靠；对于大小不受控的输入，它会把输入长度直接转化为内存压力。

2. 按行流式读取

文本文件支持迭代：

def count_errors(path):
    count = 0

    with open(path, encoding="utf-8") as f:
        for line_number, line in enumerate(f, 1):
            if "ERROR" in line:
                count += 1
                print(line_number, line.rstrip("\n"))

    return count

这一过程的关键是：

读取一部分 → 找到一行 → 处理一行 → 丢弃已处理内容 → 继续读取

它不要求把整个文件保存为一个大字符串，但单行长度仍可能很大。因此，“逐行处理”并不意味着内存恒定；它通常把峰值从“整个文件大小”降低到“缓冲区加当前记录大小”。

3. 按块读取二进制数据

复制大文件时，可以使用固定大小的字节块：

from pathlib import Path

def copy_binary(source: Path, target: Path, chunk_size: int = 1024 * 1024):
    with source.open("rb") as src, target.open("wb") as dst:
        while True:
            chunk = src.read(chunk_size)
            if not chunk:
                break
            dst.write(chunk)

read(chunk_size) 的返回值有三种重要状态：

• 非空 bytes：读取到了数据；
• 空 bytes：到达 EOF；
• 在特殊非阻塞场景下，可能抛出 BlockingIOError。

对于普通磁盘文件，空字节串通常意味着 EOF；对于网络或管道，EOF 还意味着对端关闭或写端关闭，必须结合协议理解。

4. 不要按固定字节块直接解码文本

下面的写法存在边界错误：

decoder = None

with open("input.txt", "rb") as f:
    while chunk := f.read(4):
        print(chunk.decode("utf-8"))

如果 UTF-8 的一个字符被拆在两个块之间，例如：

第一个块：e4
第二个块：b8 ad

第一次 decode() 会抛出 UnicodeDecodeError，因为 e4 只是一个不完整的 UTF-8 序列。

解决方式之一是使用增量解码器：

import codecs

def iter_text_chunks(path, chunk_size=4):
    decoder = codecs.getincrementaldecoder("utf-8")("strict")

    with open(path, "rb") as f:
        while chunk := f.read(chunk_size):
            text = decoder.decode(chunk, final=False)
            if text:
                yield text

        tail = decoder.decode(b"", final=True)
        if tail:
            yield tail

增量解码器会保存跨块的未完成字节，直到下一块到达。final=True 用于告诉解码器输入已经结束；如果此时仍有不完整序列，严格模式会报告错误。

完整使用示例：

from pathlib import Path

Path("input.txt").write_text("A中B你好", encoding="utf-8")

for piece in iter_text_chunks("input.txt", chunk_size=2):
    print(repr(piece))

输出分块边界可能类似：

'A'
'中'
'B'
'你'
'好'

分块结果的边界不应被当作字符、单词或业务记录边界。流式解码只保证字符边界，不保证应用层消息边界。

---

八、字符边界、记录边界和消息边界不是一回事

流式处理至少要区分三种边界：

字节边界
  ▼ 解码器
字符边界
  ▼ 行解析器 / 帧解析器
记录或消息边界

1. 字节边界

二进制读取可以在任意字节位置切块：

chunk = stream.read(4096)

但任意字节块不一定构成完整字符。

2. 字符边界

增量解码器可以避免把一个 UTF-8 字符拆成非法字符串，但得到的字符串片段仍可能只包含半行：

第一块：'user=alice\nus'
第二块：'er=bob\n'

3. 记录或消息边界

如果协议规定“每行是一条记录”，就必须维护一个字符串累积区：

def iter_lines_from_binary(path, chunk_size=4096):
    decoder = codecs.getincrementaldecoder("utf-8")("strict")
    pending = ""

    with open(path, "rb") as f:
        while chunk := f.read(chunk_size):
            pending += decoder.decode(chunk, final=False)

            while True:
                line, separator, rest = pending.partition("\n")
                if not separator:
                    pending = line
                    break

                yield line
                pending = rest

        pending += decoder.decode(b"", final=True)
        if pending:
            yield pending

这里的状态有两个：

1. 解码器内部可能保留不完整 UTF-8 序列；
2. pending 可能保留尚未遇到换行符的完整字符。

如果协议使用长度前缀，则不能用换行分割；如果协议使用固定长度，则应在字节层完成长度控制，再根据协议编码解码。解析边界必须由协议定义，而不是由 read() 一次返回多少字节决定。

---

九、标准输入输出、管道和 socket 的特殊性

1. 管道和 socket 没有“天然的一条消息”

对文件，read() 通常可以持续读到 EOF；对管道和 socket，数据可能分批抵达。

错误假设：

message = sock.recv(1024)
# 认为 message 就是一条完整消息

recv(1024) 的含义通常只是“最多读取 1024 字节当前可用数据”。一次返回可能：

• 少于一条消息；
• 恰好一条消息；
• 包含多条消息；
• 在消息中间结束。

因此应用协议必须定义边界，例如：

• 固定长度；
• 长度前缀；
• 分隔符；
• 明确的连接关闭。

2. 文本包装器可能预读

TextIOWrapper 为了完成解码和换行识别，可能从底层读取比调用者当前请求更多的字节，并把多余内容放在自己的缓冲区中。因此，不应在同一个底层对象上随意交替操作：

text = io.TextIOWrapper(binary_stream, encoding="utf-8")

text.readline()
binary_stream.read(10)  # 可能跳过或读取到预期之外的位置

文本层和底层二进制层应保持清晰的所有权关系。如果确实需要切换层次，应理解 flush()、seek() 和 detach() 的语义。detach() 会分离底层流，但原文本对象随后不可继续使用。(docs.python.org)

3. 非阻塞流可能返回“现在还没有数据”

当底层流是非阻塞的，文本读操作可能抛出 BlockingIOError，因为当前无法立即完成所需操作。这与 EOF 不同：

BlockingIOError：暂时没有足够数据，之后可能继续读到
b"" 或 ""：流已到达 EOF

事件循环、线程或异步框架通常会负责等待可读事件。同步代码不能把 BlockingIOError 当作文件结束。

---

十、缓冲区大小和 readinto()：减少额外分配

1. buffering 控制的不是所有层

open("data.bin", "rb", buffering=8192)

这里主要影响二进制缓冲层。文本流还包含字符解码、换行处理和自身的文本缓冲；不能简单地把 buffering=0 理解为“整个文本流完全无缓冲”。Python 文档也特别指出，TextIOWrapper 的缓冲行为与二进制缓冲不同。(docs.python.org)

通常应先使用默认缓冲。只有在已经确认 I/O 模式、系统调用次数或延迟要求后，才调整缓冲参数。

2. readinto() 可以复用调用方提供的缓冲区

对于二进制流，可以提供可写缓冲区：

buffer = bytearray(4096)

with open("data.bin", "rb") as f:
    while True:
        n = f.readinto(buffer)
        if not n:
            break

        view = memoryview(buffer)[:n]
        process(view)

readinto() 把数据写入已有的 bytearray，避免每次读取都创建新的 bytes 对象。它适合高频、固定大小的二进制处理；但如果 process() 需要长期保存数据，必须复制，因为下一次读取会覆盖同一块缓冲区。

---

十一、内存中的流：BytesIO 和 StringIO

1. BytesIO 是内存中的二进制流

import io

stream = io.BytesIO()
stream.write(b"abc")
stream.write(b"\x00\xff")

print(stream.getvalue())  # b'abc\x00\xff'

它适合：

• 测试需要文件接口的函数；
• 组装待上传的二进制内容；
• 构造压缩包、图片或协议帧；
• 在内存中模拟二进制流。

2. StringIO 是内存中的文本流

import io

stream = io.StringIO()
stream.write("第一行\n")
stream.write("第二行\n")

print(stream.getvalue())

如果使用 StringIO，数据已经是 str，不存在底层编码过程。要获得 UTF-8 字节，必须显式编码：

data = stream.getvalue().encode("utf-8")

这说明“文本流”并不一定有文件或字节层。StringIO 可以直接在内存中提供文本接口，而 TextIOWrapper 才是典型的“字节流之上的文本包装器”。(docs.python.org)

---

十二、文件写入与原子替换：I/O 正确不等于更新流程正确

文件 I/O 还要面对并发和故障路径。下面这种更新方式存在中间状态：

with open("config.json", "w", encoding="utf-8") as f:
    f.write(new_content)

w 会先截断原文件，然后逐步写入新内容。如果进程在写入中途崩溃，其他进程可能看到空文件或不完整文件。

更安全的常见流程是：

1. 在同一目录创建临时文件
2. 将完整内容写入临时文件
3. 刷新并关闭临时文件
4. 用 os.replace() 替换目标路径

示例：

import os
import tempfile
from pathlib import Path

def atomic_write_text(path: Path, text: str, encoding="utf-8"):
    path = Path(path)

    fd, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.",
        dir=path.parent,
        text=False,
    )

    try:
        with os.fdopen(fd, "w", encoding=encoding, newline="") as f:
            f.write(text)
            f.flush()
            os.fsync(f.fileno())

        os.replace(temporary_name, path)
    except BaseException:
        try:
            os.unlink(temporary_name)
        except FileNotFoundError:
            pass
        raise

这个流程解决的是“读者看到半写文件”的问题，但还需要区分几个边界：

• os.replace() 的原子性依赖源文件和目标文件位于同一文件系统语义范围内；
• 原子替换不自动解决目录项持久化和断电恢复问题；
• 它不等价于多进程协调，多个写者仍可能互相覆盖；
• 在替换前检查“文件是否存在”，再决定是否写入，可能产生竞态。

如果要求“目标不存在才创建”，直接使用 open(..., "x") 或底层独占创建语义，比“先检查再创建”更可靠。路径、目录遍历、元数据和竞态属于文件系统层问题，不能只靠 io 层的 write() 解决。

---

十三、异常、EOF 和关闭状态

1. EOF 不是异常

文本流到达 EOF 时，read() 返回空字符串：

with open("empty.txt", encoding="utf-8") as f:
    result = f.read()
    print(repr(result))  # ''

二进制流到达 EOF 时，返回空字节串：

with open("empty.bin", "rb") as f:
    result = f.read()
    print(repr(result))  # b''

但关闭后的流不是 EOF：

f = open("empty.txt", encoding="utf-8")
f.close()

f.read()

通常会抛出：

ValueError: I/O operation on closed file

代码应明确区分：

空返回值：仍是一个有效流，但没有更多数据
ValueError：流生命周期已经结束
OSError：操作系统或底层设备错误
UnicodeDecodeError：字节不能按指定编码解释
UnicodeEncodeError：字符串不能按指定编码表示

2. read() 返回多少不是协议保证

对于普通文件，缓冲层通常会尽量满足请求大小；但流接口的通用语义是“最多读取指定数量”，而不是“必定读取指定数量”。网络、管道、非阻塞设备和自定义流尤其如此。

如果业务需要精确读取 N 个字节，应显式循环：

def read_exact(stream, size):
    chunks = []
    remaining = size

    while remaining:
        chunk = stream.read(remaining)
        if not chunk:
            raise EOFError(
                f"expected {size} bytes, got {size - remaining}"
            )
        chunks.append(chunk)
        remaining -= len(chunk)

    return b"".join(chunks)

该函数的前提是流最终会提供数据或到达 EOF；如果流可能暂时不可读，还需要处理阻塞、超时或 BlockingIOError。

---

十四、一个完整的流式文本处理示例

下面实现一个按行统计日志中错误数量的函数。它具备几个明确性质：

• 显式指定 UTF-8；
• 不一次性读取整个文件；
• 逐行处理；
• 保留行号；
• 严格处理编码错误；
• 通过上下文管理器关闭文件。

from pathlib import Path

def count_error_lines(path: Path) -> tuple[int, int]:
    total_lines = 0
    error_lines = 0

    with path.open(
        "r",
        encoding="utf-8",
        errors="strict",
        newline=None,
    ) as stream:
        for line in stream:
            total_lines += 1
            if "ERROR" in line:
                error_lines += 1

    return total_lines, error_lines


if __name__ == "__main__":
    path = Path("app.log")
    path.write_text(
        "INFO started\n"
        "ERROR failed to connect\n"
        "INFO retrying\n"
        "ERROR timeout\n",
        encoding="utf-8",
    )

    print(count_error_lines(path))

输出：

(4, 2)

每一步的因果关系是：

1. path.open() 创建文本流；
2. encoding="utf-8" 决定字节到 str 的解码规则；
3. newline=None 将常见换行形式规范化为 \n；
4. 迭代器逐步从底层流读取数据；
5. 每次循环得到一行 str；
6. 处理完成后，下一行覆盖前一行的引用；
7. 离开 with 块时关闭流。

这个实现仍然受单行长度影响。如果日志格式允许无限长行，应该改用明确的记录协议，或者在二进制层实现有上限的分隔符解析。

---

十五、常见误解与对应诊断

误解一：UTF-8 是 Python 字符串的内部类型

不准确。Python 的 str 是 Unicode 文本对象；UTF-8 是一种把文本编码成字节的方式。只有在进入文件、网络或其他二进制边界时，才需要选择具体编码。

诊断：

text = "中"

print(type(text))                    # str
print(type(text.encode("utf-8")))     # bytes

误解二：len(text) 就是文件大小

不准确。文件大小通常以字节计，而 len(text) 统计字符串中的 Unicode 代码点数量：

text = "A中"

print(len(text))                         # 2
print(len(text.encode("utf-8")))          # 4

应根据问题选择：

len(text)                    # 文本对象的代码点数量
len(text.encode("utf-8"))     # UTF-8 编码后的字节数量
path.stat().st_size           # 文件系统中的文件大小，单位为字节

误解三：flush() 等于已经安全落盘

flush() 主要处理 Python I/O 层和底层流之间的缓冲；要讨论断电后的持久性，还要考虑操作系统、文件系统和存储设备。需要更强持久化语义时，通常还要使用底层文件描述符上的 os.fsync()，并设计完整的临时文件和替换流程。

误解四：一次 read() 对应一次消息

对文件记录也许碰巧如此，对 socket 和管道则没有这种保证。消息边界必须由协议定义，再通过循环、缓冲和解析器恢复。

误解五：把文本流和二进制流混用可以节省一次转换

如果文本层已经预读或缓存，直接操作其底层缓冲可能使当前位置和数据顺序难以推断。除非明确掌握 flush()、seek()、解码器状态和缓冲区状态，否则应选择单一抽象层完成一段连续操作。

---

十六、如何选择 I/O 方式

可以按数据边界选择接口：

场景	推荐接口	原因
UTF-8 配置文件	open(..., encoding="utf-8")	在文本边界明确编码
图片、压缩包、加密数据	open(..., "rb"/"wb")	保留原始字节
小型文本文件	Path.read_text()	简洁，自动关闭
小型二进制文件	Path.read_bytes()	直接得到完整 bytes
大型文本日志	文本流迭代或逐块读取	限制内存占用
大型二进制文件	read(chunk_size) 或 readinto()	固定内存处理
网络消息	二进制流加协议解析	明确消息边界
内存中的文本拼装	io.StringIO	不需要文件系统
内存中的二进制拼装	io.BytesIO	提供文件式二进制接口
需要低级控制	FileIO 或自定义 RawIOBase	需要自行处理部分读写
配置文件整体替换	临时文件加 os.replace()	避免读者看到半成品

最终可以用一条数据流规则概括：

外部字节
  → 二进制边界
  → 解码器
  → Python str
  → 记录解析
  → 业务对象

反向写出时则是：

业务对象
  → Python str
  → 编码器
  → 二进制缓冲
  → 操作系统文件或网络端点

只要明确每一层负责什么，就能判断问题属于哪一类：

• str 和 bytes 类型错误：文本/二进制边界错误；
• UnicodeDecodeError：编码协议或输入字节错误；
• 文件大小与字符串长度不一致：把字符数和字节数混淆；
• 数据迟迟不可见：缓冲或刷新时机问题；
• 读到半条消息：把流误认为消息队列；
• 文件出现半成品：更新流程缺少临时文件和原子替换；
• 随机定位结果异常：把文本位置当成字节偏移。

Python I/O 的核心不是记住若干个 open() 参数，而是始终回答三个问题：当前对象处理的是字符还是字节，数据边界由谁定义，当前数据究竟停留在哪一层。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 文件系统与 pathlib：路径、遍历、元数据、原子替换和竞态
• 下一篇：Python 结构化数据：JSON、CSV、TOML、Schema 与精度边界
• 延伸：Python 字符串、bytes 与 Unicode：编码、解码和文本边界

官方资料

• Python 3.14 Standard Library
• Python HOWTOs

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。