
io编程怎么用:从准备到完成的实操步骤
“IO 编程”通常指程序与外部数据源之间的输入、输出操作,例如读取文件、写入日志、处理内存中的文本或字节数据。在 Python 中,日常文件操作主要使用内置的 open();需要在内存中模拟文件、编写通用流处理代码时,则会用到标准库 io。
本文以 Python 3 为例,介绍从选择数据类型、打开流、读写数据到处理异常和安全边界的完整做法。示例不依赖第三方库,保存为 .py 文件即可运行。重点不是记住某个函数,而是先判断数据是文本还是字节,再选择正确的流和参数。
一、先判断:你要处理的是什么 IO
IO 可以按数据来源和数据形式来理解。
- 文件 IO:读取或写入磁盘上的文件,通常使用
open()。 - 内存 IO:把内存对象当作文件使用,适合测试、拼装内容或调用要求文件接口的库,使用
io.StringIO或io.BytesIO。 - 文本 IO:处理有字符编码的内容,例如 UTF-8 的 JSON、CSV、配置文件。
- 二进制 IO:处理原始字节,例如图片、压缩包、音频、网络协议数据。
文件描述符、套接字、管道和终端也属于 IO 场景,但它们的生命周期、阻塞行为和错误处理可能不同。本文聚焦于 Python 的常规文件流和内存流,不把它们等同于异步网络编程。
在开始前,确认 Python 版本,并尽量使用当前仍受支持的 Python 3 版本。io 是 Python 标准库,无需安装。可以在终端运行 python --version 检查环境;有些系统的命令名是 python3。
二、选对流:文本流与二进制流
Python 的 IO 抽象中,文本流负责字符与字节间的编码、解码;二进制流直接读写字节。两者不能混用:文本流的 read() 返回 str,二进制流的 read() 返回 bytes。写入时也分别需要 str 和字节类对象。
| 需求 | 典型打开方式 | 读取结果 |
|---|---|---|
| 文本文件 | open(path, "r", encoding="utf-8") | str |
| 二进制文件 | open(path, "rb") | bytes |
| 内存文本流 | io.StringIO() | str |
| 内存二进制流 | io.BytesIO() | bytes |
文本文件应明确指定编码,避免程序依赖操作系统的默认编码。对常见的跨平台文本文件,encoding="utf-8" 是实用选择。二进制模式不接受 encoding 参数,因为它不负责字符解码。
三、第一步:用上下文管理器打开文件
文件打开后会占用系统资源。使用 with 语句,即使读取或写入过程中发生异常,离开代码块时也会关闭文件。通常不需要手动调用 close()。
from pathlib import Path
path = Path("notes.txt")
with path.open("w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
with path.open("r", encoding="utf-8") as f:
content = f.read()
print(content)
也可以直接使用内置函数 open()。Path.open() 和 open() 的常见用法相近。上例使用写入模式 "w":如果文件不存在就创建,如果已经存在则会截断原有内容。若不能覆盖旧文件,不要误用它。
四、第二步:理解打开模式,避免误覆盖
常用模式如下。模式字符串由读写、追加、创建和二进制等标记组合而成。
"r":读取;文件不存在时抛出FileNotFoundError。"w":写入;文件不存在时创建,存在时先截断内容。"a":追加到文件末尾;文件不存在时创建。"x":独占创建;若路径对应的文件已存在则失败,适合避免意外覆盖。"r+":读取和写入,不会自动截断;需要留意文件指针位置。- 在模式中加入
"b",例如"rb"、"wb",即可使用二进制模式。
模式的具体组合需遵循 open() 的规则。遇到权限不足、路径无效、磁盘空间不足等情况,操作也可能失败;模式本身不会替程序处理这些问题。写重要数据前,先确认目标文件是否允许覆盖。
五、第三步:按需读取,而不是一律读完整个文件
无参数调用 read() 会读取剩余的全部内容。对于大小有限、来源可信的配置文件或小型数据,这种写法简洁;对于大文件或用户可控输入,应考虑逐行、分块处理。read(size) 中的 size 表示读取数量,但文本流按字符计数,二进制流按字节计数;它不是对进程总内存使用量的硬性限制。
from pathlib import Path
path = Path("events.log")
# 逐行处理文本文件,内存中不必同时保存所有行
with path.open("r", encoding="utf-8") as f:
for line in f:
line = line.rstrip("\n")
if "ERROR" in line:
print(line)
# 分块处理二进制文件
chunk_size = 64 * 1024
with Path("input.bin").open("rb") as src:
with Path("copy.bin").open("wb") as dst:
while True:
chunk = src.read(chunk_size)
if not chunk:
break
dst.write(chunk)
分块大小是应用选择,不存在适用于所有场景的固定值。较大的块通常减少调用次数,但会增加单次缓冲的内存;较小的块则相反。上例控制了每次显式读取的大小,但 Python、操作系统和底层设备仍可能有自己的缓冲,因此不能把它当成严格的进程内存上限。
文本逐行迭代适合行结构数据,但并不代表每一行都很小。如果输入中可能有极长的一行,逐行读取仍可能占用大量内存。对不可信输入,应在协议或业务层定义并校验合理的数据长度。
六、第四步:写文本、写二进制数据
文本流的 write() 接收字符串,返回写入的字符数;二进制流接收字节数据,返回写入的字节数。返回值不应被误认为一定等于最终落盘的字节数,特别是文本内容还要经过编码转换。
from pathlib import Path
# 写文本
with Path("message.txt").open("w", encoding="utf-8") as f:
count = f.write("你好,IO\n")
print("写入字符数:", count)
# 写二进制数据
payload = bytes([0, 1, 2, 255])
with Path("payload.bin").open("wb") as f:
written = f.write(payload)
print("写入字节数:", written)
如果数据原本就是文件字节、压缩内容或协议数据,不要先用文本编码去解释它。反过来,如果手上是正常文本,也不要把它随意转换成某种编码后再用二进制方式拼接,除非你明确知道目标格式要求什么字节布局。
七、第五步:掌握 StringIO 和 BytesIO
io.StringIO 在内存中提供类似文本文件的接口,io.BytesIO 则提供类似二进制文件的接口。它们适合单元测试、生成小型内容、把数据传给需要文件对象的函数等场景。由于内容保存在内存中,不适合未经评估就承载大型数据。
from io import StringIO, BytesIO
text_stream = StringIO()
text_stream.write("name,score\n")
text_stream.write("Ada,98\n")
# getvalue() 返回当前全部内容,但不会改变流的位置
print(text_stream.getvalue())
# seek(0) 把当前位置移到开头,随后可以像读取文件一样读取
text_stream.seek(0)
print(text_stream.read())
binary_stream = BytesIO()
binary_stream.write(b"\x00\x01\x02")
binary_stream.seek(0)
print(binary_stream.read())
流有“当前位置”。写入后立即读取时,位置通常位于刚写入内容之后,所以可能读到空结果;调用 seek(0) 可以回到开头。getvalue() 会返回全部内容,而 read() 从当前位置读取,两者行为不同。
在需要生成文本 CSV 的场景中,StringIO 也可以交给标准库的 csv 模块:
import csv
from io import StringIO
buffer = StringIO(newline="")
writer = csv.writer(buffer)
writer.writerow(["name", "score"])
writer.writerow(["Ada", 98])
csv_text = buffer.getvalue()
print(csv_text)
这里的 newline="" 是 csv 模块推荐的文本文件打开方式之一,用于让模块自行处理换行。对于普通文本读取,是否指定 newline 要结合文件格式和所需的换行处理来决定。
八、缓冲、刷新与文件指针
Python 的文件对象通常带有缓冲。写入时,数据可能先进入缓冲区,再由运行时交给操作系统;关闭文件时,缓冲会被刷新。flush() 可以请求将 Python 层缓冲的数据送出,但它本身不等同于保证数据已经可靠写入物理介质。若程序需要更强的持久性保证,还要理解操作系统和存储设备的语义,不能只依赖 flush()。
seek() 用于移动文件指针,tell() 返回当前位置。对二进制流,偏移量通常按字节理解;文本流涉及解码与换行转换,随意按字节跳转容易出错,应优先使用文本流允许的定位方式,或改用二进制模式并自行处理编码。
from io import BytesIO
stream = BytesIO(b"abcdef")
print(stream.tell()) # 0
stream.seek(2)
print(stream.read(2)) # b"cd"
print(stream.tell()) # 4
对于一般文件读取,顺序遍历通常比频繁跳转更简单可靠。若确实需要随机访问,先确认数据格式是否支持按字节定位。
九、异常处理:区分失败原因,避免吞掉错误
文件操作常见异常包括 FileNotFoundError、PermissionError、IsADirectoryError 和更一般的 OSError。编码不匹配时,读取文本还可能抛出 UnicodeDecodeError。不要用一个空的 except: 把所有错误吞掉,否则程序可能在数据未读完或未写成功时继续运行。
from pathlib import Path
path = Path("settings.txt")
try:
with path.open("r", encoding="utf-8") as f:
settings = f.read()
except FileNotFoundError:
print("配置文件不存在")
except PermissionError:
print("没有读取权限")
except UnicodeDecodeError as exc:
print("文件不是预期的 UTF-8 文本:", exc)
except OSError as exc:
print("文件操作失败:", exc)
是否捕获异常取决于程序的责任边界。命令行工具可以给出清楚的错误信息并退出;库函数通常应让调用方决定如何恢复。无论哪种方式,都应避免报告“成功”但实际上只完成了部分读写。
十、安全边界:路径、覆盖、输入大小与敏感数据
1. 不要直接信任外部提供的路径
如果程序允许用户指定文件路径,攻击者可能借助 ../ 等路径片段访问预期目录之外的文件。可以先解析路径并检查它是否位于允许的根目录内:
from pathlib import Path
root = Path("uploads").resolve()
user_name = "report.txt"
target = (root / user_name).resolve()
if not target.is_relative_to(root):
raise ValueError("路径不在允许目录内")
with target.open("wb") as f:
f.write(b"example")
Path.is_relative_to() 在较新的 Python 版本中提供;若要支持更旧版本,应采用该版本对应的路径检查方案。路径检查需要结合应用的威胁模型:在存在并发修改、符号链接或复杂权限边界的环境中,仅靠字符串拼接或一次路径校验未必能消除竞态风险。对上传服务等敏感系统,应使用受控目录、最小权限和适合平台的安全文件创建方式。
2. 避免意外覆盖与部分写入
对重要文件,不要不加判断地使用 "w"。使用 "x" 可以在目标已存在时拒绝创建,但它不解决所有并发或业务一致性问题。需要替换配置文件时,常见做法是在同一文件系统的临时文件中写完,再用 os.replace() 替换目标;替换操作的具体原子性和持久性保证受平台及文件系统影响。临时文件应妥善清理,并根据数据重要性评估是否还需要同步写入。
3. 对不可信内容设置业务层限制
调用 read() 读取全部内容会让内容占用内存。对于网络输入、用户上传文件或可能很大的文件,应使用分块读取,并在应用层设置允许的总字节数、行长度或记录数量。注意,文本流的读取数量按字符计,而不是按 UTF-8 字节数计;如果限制必须以字节为准,应在二进制层处理并验证。
4. 不要把日志当作安全存储
写日志前应考虑是否会泄露密码、令牌、个人信息或密钥。文件权限、日志轮转、保留时间和备份方式都可能影响数据安全。关闭文件和调用 flush() 不是加密,也不等于安全擦除;如果数据需要保密,应使用适当的访问控制和加密方案。
十一、常见错误与修正方式
- 忘记指定文本编码:在不同系统上可能得到不同结果。打开文本文件时明确写出
encoding。 - 把二进制数据当作字符串:图片、压缩包等应使用
"rb"或"wb",不要随意解码。 - 误用
"w"覆盖旧文件:先确认覆盖意图;要求文件不存在时才创建,可考虑"x"。 - 读完整个大文件:改为迭代行或分块读取,并为极长行和总数据量设置业务限制。
- 写后立刻读却得到空内容:检查当前位置,必要时调用
seek(0)。 - 把流位置当作字节偏移:文本流存在编码和换行处理,按字节跳转时优先使用二进制流。
- 异常被静默忽略:捕获具体错误并给出可诊断的信息,不要用裸
except:掩盖失败。 - 认为分块就绝对不会占用太多内存:分块只限制单次显式读取量;业务缓存、超长记录和底层缓冲仍需考虑。
十二、完整示例:安全地分块复制二进制文件
下面的程序接收源路径和目标路径,按块复制文件,并避免在目标文件已存在时覆盖它。它适用于普通二进制文件复制示例,不提供完整的并发安全、权限隔离或恶意路径防护;生产环境需要根据调用来源补充校验。
from pathlib import Path
import argparse
def copy_file(source: Path, destination: Path, chunk_size: int = 64 * 1024) -> None:
if chunk_size <= 0:
raise ValueError("chunk_size 必须大于 0")
with source.open("rb") as src:
# 模式 x:目标存在时抛出 FileExistsError,不覆盖
with destination.open("xb") as dst:
while True:
chunk = src.read(chunk_size)
if not chunk:
break
dst.write(chunk)
def main() -> None:
parser = argparse.ArgumentParser(description="分块复制二进制文件")
parser.add_argument("source", type=Path, help="源文件路径")
parser.add_argument("destination", type=Path, help="目标文件路径")
args = parser.parse_args()
try:
copy_file(args.source, args.destination)
except FileNotFoundError as exc:
parser.error(f"文件不存在:{exc.filename}")
except FileExistsError:
parser.error("目标文件已存在;为避免覆盖,复制已取消")
except OSError as exc:
parser.error(f"文件操作失败:{exc}")
if __name__ == "__main__":
main()
将代码保存为 copy_binary.py,在终端运行:
python copy_binary.py input.bin output.bin
示例中的 chunk_size 由程序设定为 64 KiB,并且函数只检查它大于零;这不是 Python 对文件大小或内存用量规定的上限。目标以 "xb" 打开,若文件已存在,创建会失败而不是覆盖。若复制中途发生错误,目标文件可能已经包含部分数据;正式程序可以在受控目录创建临时文件,复制成功后再替换目标,并设计失败清理逻辑。
十三、异步 IO 不是 io 模块的另一种打开模式
io 主要提供同步流抽象。异步文件、网络和数据库 IO 则由相应的异步 API、事件循环或第三方库处理,不能只把 open() 改个参数就变成异步。对于简单脚本和本地文件,普通同步 IO 往往更清晰;只有在程序结构、并发需求和所用库都支持异步时,才需要引入异步方案。
结语:按数据类型与风险选择做法
使用 Python IO 时,可以按这个顺序检查:先确认数据是文本还是字节;再选文件流或内存流;打开文本文件时明确编码;选择不会意外覆盖数据的模式;根据数据规模决定整体读取、逐行读取还是分块读取;最后处理异常、路径和输入大小等安全边界。
小文件和临时内容可以用简洁的 with open();大型二进制数据适合分块处理;测试和小型内存内容适合 StringIO、BytesIO。掌握这些边界后,IO 代码就不只是“能读能写”,也更容易维护、排错和安全地投入实际使用。









