io编程怎么用:从准备到完成的实操步骤

教程1天前发布 jdkjadmin
0 00
io编程怎么用:从准备到完成的实操步骤

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 代码就不只是“能读能写”,也更容易维护、排错和安全地投入实际使用。

© 版权声明

相关文章