当前位置： 首页 > news >正文

Python I/O 库【输入输出】全面详解

news 2025/7/14 8:36:29

Python 的 I/O（输入/输出）系统是处理数据流的核心机制，涵盖文件操作、内存流、标准输入输出、序列化等。下面将从基础到高级进行全面讲解，包含详细原理和示例。

一、文件 I/O 深度解析
1. 文件打开模式详解
模式描述文件存在文件不存在指针位置
'r'只读打开报错(FileNotFoundError)开头
'r+'读写打开报错开头
'w'只写清空创建开头
'w+'读写清空创建开头
'a'追加打开创建末尾
'a+'读写追加打开创建末尾
'x'排他创建报错(FileExistsError)创建开头
'b'二进制模式（需组合使用）
't'文本模式（默认）

重要特性：

  • 二进制模式(b)与文本模式(t)互斥
  • +号启用读写功能，但行为因主模式而异
  • Windows 系统中换行符自动转换（文本模式）
2. 文件操作全方法
with open('data.txt', 'r+', encoding='utf-8') as f:# 读取操作print(f.read(10))        # 读取前10个字符print(f.readline())      # 读取下一行print(f.readlines())     # 剩余所有行 -> 列表# 写入操作f.write("新内容\n")      # 在当前位置写入f.writelines(["行1\n", "行2\n"])# 指针控制f.seek(5)                # 移动到第5字节print(f.tell())          # 输出当前位置 -> 5f.seek(0, 2)             # 移动到文件末尾# 缓冲区控制f.flush()                # 强制写入磁盘# 文件截断f.truncate(20)           # 截断文件到20字节
3. 二进制文件操作
# 图像复制
with open('source.jpg', 'rb') as src, open('copy.jpg', 'wb') as dst:while True:chunk = src.read(4096)  # 分块读取(4KB)if not chunk:breakdst.write(chunk)
二、上下文管理器原理

with 语句实现原理：

class ManagedFile:def __init__(self, filename, mode):self.filename = filenameself.mode = modedef __enter__(self):self.file = open(self.filename, self.mode)return self.filedef __exit__(self, exc_type, exc_val, exc_tb):if self.file:self.file.close()# 可处理异常if exc_type:print(f"异常发生: {exc_val}")return True  # 抑制异常# 使用自定义上下文管理器
with ManagedFile('data.txt', 'w') as f:f.write("上下文管理")
三、标准 I/O 流详解
import sys# 重定向标准输出
with open('output.log', 'w') as f:sys.stdout = fprint("此内容写入文件")sys.stdout = sys.__stdout__  # 恢复# 非阻塞输入检查
import select
while True:# 检查标准输入是否有数据rlist, _, _ = select.select([sys.stdin], [], [], 0.1)if rlist:line = sys.stdin.readline().strip()if line == 'exit':breakprint(f"收到: {line}")else:print("等待输入...")
四、内存流高级应用
1. StringIO 复杂操作
from io import StringIO, BytesIO# 文本流
stream = StringIO()
stream.write("初始内容")
stream.seek(0)# 插入内容到指定位置
content = stream.getvalue()
new_content = content[:3] + "插入" + content[3:]
stream = StringIO(new_content)# 流转换
binary_stream = BytesIO(stream.getvalue().encode('utf-8'))
2. BytesIO 二进制处理
# 创建PNG文件头
png_header = bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A])
img_data = BytesIO()
img_data.write(png_header)
img_data.write(b"\x00\x00\x00\x0D")  # IHDR长度
img_data.write(b"IHDR")              # 块类型# 验证文件头
img_data.seek(0)
if img_data.read(8) == png_header:print("有效PNG文件")
五、序列化高级技巧
1. Pickle 安全与优化
import pickle
import functoolsclass ComplexObject:def __init__(self, data):self.data = data# 自定义序列化def __getstate__(self):return {"safe_data": self.data * 2}# 自定义反序列化def __setstate__(self, state):self.data = state["safe_data"] / 2# 高效序列化协议
obj = ComplexObject(42)
with open('data.pkl', 'wb') as f:# 使用协议版本5（Python 3.8+）pickle.dump(obj, f, protocol=5, buffer_callback=functools.partial(f.write))# 增量加载
unpickler = pickle.Unpickler(open('data.pkl', 'rb'))
while True:try:item = unpickler.load()except EOFError:break
2. JSON 高级处理
import json
from datetime import datetime
import numpy as npclass CustomEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, datetime):return obj.isoformat()if isinstance(obj, np.ndarray):return obj.tolist()return super().default(obj)data = {"time": datetime.now(),"matrix": np.array([[1, 2], [3, 4]])
}# 自定义序列化
json_str = json.dumps(data, cls=CustomEncoder, indent=2)# 自定义反序列化
def object_hook(dct):if 'time' in dct:dct['time'] = datetime.fromisoformat(dct['time'])return dctdata = json.loads(json_str, object_hook=object_hook)
六、高级 I/O 控制
1. 缓冲机制深度
缓冲类型设置方式刷新条件适用场景
无缓冲buffering=0立即写入实时日志
行缓冲buffering=1遇到换行符终端交互
全缓冲buffering>1缓冲区满文件操作
# 行缓冲实时监控
with open('log.txt', 'w', buffering=1) as f:  # 行缓冲for i in range(5):f.write(f"日志条目 {i}\n")input("按回车继续")  # 每次写入后立即刷新
2. 内存映射文件
import mmapwith open('large.bin', 'r+b') as f:# 创建内存映射mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_WRITE)# 随机访问print(mm[1000:1020])  # 读取字节# 修改内容mm[500:510] = b'X' * 10# 搜索内容pos = mm.find(b'PATTERN')if pos != -1:mm.seek(pos)mm.write(b'REPLACEMENT')mm.close()  # 必须手动关闭
七、文件系统高级操作
1. 目录遍历优化
import os
from pathlib import Path# 高效遍历大目录
def scan_large_dir(path):with os.scandir(path) as entries:for entry in entries:if entry.is_dir():print(f"目录: {entry.name}")elif entry.is_file():print(f"文件: {entry.name} ({entry.stat().st_size}字节)")# 使用pathlib递归处理
root = Path('project')
for file in root.glob('**/*.py'):  # 所有Python文件print(f"Python文件: {file.relative_to(root)}")
2. 文件锁机制
import fcntldef safe_write(path, content):with open(path, 'a') as f:# 获取排他锁fcntl.flock(f, fcntl.LOCK_EX)try:f.write(content + "\n")finally:# 释放锁fcntl.flock(f, fcntl.LOCK_UN)# 多进程安全写入
from multiprocessing import Pool
with Pool(4) as p:p.map(lambda x: safe_write('data.txt', f"进程{x}"), range(10))
八、异步 I/O (asyncio)
import asyncio
import aiofilesasync def async_file_ops():# 异步写入async with aiofiles.open('async.txt', 'w') as f:await f.write("异步写入内容\n")await f.flush()# 异步读取async with aiofiles.open('async.txt', 'r') as f:content = await f.read()print(f"读取内容: {content}")# 运行异步任务
asyncio.run(async_file_ops())# 高性能日志记录器
class AsyncLogger:def __init__(self, filename):self.filename = filenameself.queue = asyncio.Queue()self.task = asyncio.create_task(self._writer())async def log(self, message):await self.queue.put(f"{asyncio.get_event_loop().time()}: {message}\n")async def _writer(self):async with aiofiles.open(self.filename, 'a') as f:while True:message = await self.queue.get()await f.write(message)await f.flush()
九、性能优化策略
  1. 大文件处理黄金法则
# 分块读取处理
CHUNK_SIZE = 16 * 1024  # 16KB
with open('huge.log', 'r') as f:while True:chunk = f.read(CHUNK_SIZE)if not chunk:breakprocess(chunk)# 零拷贝技术 (Linux)
import os
def zero_copy(source, dest):os.sendfile(dest.fileno(), source.fileno(), None, os.path.getsize(source))
  1. 缓冲区优化
# 自定义缓冲区
class BufferedWriter:def __init__(self, file, buffer_size=8192):self.file = fileself.buffer = bytearray(buffer_size)self.pos = 0def write(self, data):data = memoryview(data)while data:n = min(len(data), len(self.buffer) - self.pos)self.buffer[self.pos:self.pos+n] = data[:n]self.pos += ndata = data[n:]if self.pos == len(self.buffer):self.flush()def flush(self):if self.pos > 0:self.file.write(self.buffer[:self.pos])self.pos = 0def close(self):self.flush()self.file.close()# 使用自定义缓冲
with open('optimized.bin', 'wb') as raw_f:with BufferedWriter(raw_f, 65536) as buf_f:  # 64KB缓冲for _ in range(10000):buf_f.write(b'x' * 1024)  # 1KB写入

最佳实践总结

  1. 资源管理

    • 始终使用 with 语句确保资源释放
    • 长时间打开的文件定期 flush()
    • 使用 try/finally 作为 with 的备选

  2. 编码处理

    • 显式指定编码：encoding='utf-8'
    • 处理编码错误：errors='replace' 或 errors='ignore'
    • 二进制数据坚持使用 b 模式

  3. 性能关键点

    • 大文件使用迭代处理：for line in file
    • 避免频繁小量写入，使用缓冲
    • 减少系统调用次数（批量操作）

  4. 高级技巧

    • 内存映射处理超大文件
    • 异步 I/O 处理高并发
    • 文件锁保证多进程安全

  5. 调试技巧

    • 使用 file.tell() 追踪指针位置
    • 检查 os.stat() 获取文件状态
    • 监控 io 模块的 DEFAULT_BUFFER_SIZE

掌握这些 I/O 技术，可高效处理从简单文本到 TB 级数据集的各类场景。

http://www.dtcms.com/a/277675.html

相关文章：

  • JavaScript加强篇——第九章 正则表达式高级应用（终）
  • Python __main__ 全面深度解析
  • C++ 右值引用和移动语义的应用场景
  • python的平安驾校管理系统
  • Python自动化：每日销售数据可视化
  • Linux-线程控制
  • System.getenv()拿不到你配置的环境变量
  • 【Mysql作业】
  • OSPF协议特性
  • kettle从入门到精通 第九十七课 ETL之kettle kettle资源仓库的5种方式
  • Linux修炼：开发工具
  • linux-shell脚本
  • 学习环形数组ringbuffer和缓存管理buffer_manager_struct的一些思考
  • k8s:0/1 nodes are available: pod has unbound immediate PersistentVolumeClaims.
  • CSS个人笔记分享【仅供学习交流】
  • 深度学习图像分类数据集—角膜溃疡识别分类
  • INA226 数据手册解读
  • CCS-MSPM0G3507-6-模块篇-OLED的移植
  • Leetcode 3614. Process String with Special Operations II
  • 【Vue】浏览器缓存 sessionStorage、localStorage、Cookie
  • XXL-TOOL v1.5.0 发布 | Java工具类库
  • https交互原理
  • 010_学习资源与社区支持
  • cs285学习笔记（一）：课程总览
  • 融合开源AI大模型与MarTech：AI智能名片与S2B2C商城小程序源码赋能数字化营销新生态
  • Boost.Asio 中 io_context 类 post 和 dispatch的区别
  • 启动Tomcat报错：A child container failed during start
  • MCP 服务开发到发布
  • 更换docker工作目录
  • MongoDB对接SpringBoot【大数据存储】