Python IO优化案例：文件读写优化
使用缓冲区优化
# 不推荐：逐行读取，频繁IO
def slow_read(filepath):
with open(filepath, 'r') as f:
lines = []
for line in f:
lines.append(line.strip())
return lines
# 推荐：设置大缓冲区
def fast_read(filepath):
with open(filepath, 'r', buffering=65536) as f: # 64KB缓冲区
return [line.strip() for line in f]
# 使用内存映射文件（大文件）
import mmap
def mmap_read(filepath):
with open(filepath, 'r+b') as f:
with mmap.mmap(f.fileno(), 0) as mm:
return mm.read().decode().splitlines()
批量写入优化
import time
# 不推荐：频繁写入
def slow_write(data_list, filepath):
with open(filepath, 'w') as f:
for data in data_list:
f.write(f"{data}\n") # 每次写入都触发IO
# 推荐：批量写入
def fast_write(data_list, filepath):
with open(filepath, 'w') as f:
# 一次性构建字符串，减少IO次数
content = '\n'.join(str(d) for d in data_list)
f.write(content)
# 使用StringIO作为缓冲区
from io import StringIO
def buffer_write(data_list, filepath):
buffer = StringIO()
for data in data_list:
buffer.write(f"{data}\n")
with open(filepath, 'w') as f:
f.write(buffer.getvalue())
buffer.close()
使用更快的序列化方式
import json
import pickle
import msgpack
# 对比不同序列化方式
def compare_serialization(data, filepath):
import time
# JSON
start = time.time()
with open(f"{filepath}.json", 'w') as f:
json.dump(data, f, default=str)
print(f"JSON: {time.time() - start:.4f}s")
# Pickle（二进制更快）
start = time.time()
with open(f"{filepath}.pkl", 'wb') as f:
pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
print(f"Pickle: {time.time() - start:.4f}s")
# MessagePack（更快更小）
try:
start = time.time()
with open(f"{filepath}.msgpack", 'wb') as f:
f.write(msgpack.packb(data))
print(f"MsgPack: {time.time() - start:.4f}s")
except:
print("MsgPack not available")
并行读写优化
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import os
def parallel_file_read(file_list):
"""并行读取多个文件"""
def read_file(filepath):
with open(filepath, 'r') as f:
return f.read()
with ThreadPoolExecutor(max_workers=4) as executor:
results = list(executor.map(read_file, file_list))
return results
def process_large_file_chunks(filepath, chunk_size=65536):
"""分块处理大文件"""
with open(filepath, 'r') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
# 使用内存映射并行处理
import numpy as np
def parallel_numpy_array():
"""并行处理numpy数组到文件"""
# 创建大型numpy数组
large_array = np.random.rand(1000000)
# 保存为二进制格式（最快的I/O）
np.save('large_array.npy', large_array)
# 使用内存映射读取
memmap_array = np.load('large_array.npy', mmap_mode='r')
return memmap_array
避免不必要的转换
# 不推荐：多次转换
def slow_conversion(filepath):
with open(filepath, 'r') as f:
data = f.read() # 字符串
lines = data.split('\n') # 列表
# 处理数据...
# 推荐：直接处理
def fast_processing(filepath):
with open(filepath, 'r') as f:
# 直接在流上操作，避免中间转换
for line in f:
# 处理每一行
pass
完整优化示例
import os
import time
from contextlib import contextmanager
class OptimizedFileHandler:
"""优化的文件处理器"""
def __init__(self, filepath, buffer_size=65536):
self.filepath = filepath
self.buffer_size = buffer_size
@contextmanager
def open(self, mode='r'):
"""使用上下文管理器打开文件"""
f = open(self.filepath, mode, buffering=self.buffer_size)
try:
yield f
finally:
f.close()
def read_large_file(self, chunk_size=65536):
"""分块读取大文件"""
with self.open('rb') as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
def write_batch(self, data_list):
"""批量写入数据"""
# 预分配buffer大小
total_size = sum(len(str(d)) + 1 for d in data_list)
with self.open('w', buffering=total_size) as f:
f.writelines(f"{d}\n" for d in data_list)
def read_selected_lines(self, line_numbers):
"""使用seek读取指定行"""
results = []
with self.open('r') as f:
# 创建行偏移量索引（仅对大文件有用）
positions = self._build_line_index(f)
for line_num in line_numbers:
if line_num in positions:
f.seek(positions[line_num])
results.append(f.readline().strip())
return results
def _build_line_index(self, f):
"""构建行偏移量索引"""
positions = {}
pos = 0
for i, line in enumerate(f, 1):
positions[i] = pos
pos += len(line)
return positions
# 性能比较函数
def performance_comparison():
"""比较不同读写方式的性能"""
import random
import string
# 生成测试数据
test_data = [''.join(random.choices(string.ascii_letters, k=50))
for _ in range(100000)]
# 测试写入性能
print("写入性能测试:")
# 方式1：逐行写入
start = time.time()
with open('test_slow.txt', 'w') as f:
for item in test_data:
f.write(f"{item}\n")
print(f" 逐行写入: {time.time() - start:.4f}s")
# 方式2：批量写入
start = time.time()
with open('test_fast.txt', 'w') as f:
f.write('\n'.join(test_data)) # 注意：少了最后的换行
print(f" 批量写入: {time.time() - start:.4f}s")
# 方式3：使用优化处理器
handler = OptimizedFileHandler('test_optimized.txt')
start = time.time()
handler.write_batch(test_data)
print(f" 优化写入: {time.time() - start:.4f}s")
# 测试读取性能
print("\n读取性能测试:")
# 方式1：逐行读取
start = time.time()
with open('test_fast.txt', 'r') as f:
lines = [line.strip() for line in f]
print(f" 逐行读取: {time.time() - start:.4f}s")
# 方式2：全部读取
start = time.time()
with open('test_fast.txt', 'r') as f:
content = f.read()
lines = content.split('\n')[:-1] # 去掉最后一个空元素
print(f" 全部读取: {time.time() - start:.4f}s")
# 方式3：使用mmap（对于大文件更有优势）
import mmap
start = time.time()
with open('test_fast.txt', 'r+b') as f:
with mmap.mmap(f.fileno(), 0) as mm:
content = mm.read().decode()
lines = content.split('\n')[:-1]
print(f" Mmap读取: {time.time() - start:.4f}s")
# 清理测试文件
for f in ['test_slow.txt', 'test_fast.txt', 'test_optimized.txt']:
if os.path.exists(f):
os.remove(f)
if __name__ == "__main__":
performance_comparison()
- 增大缓冲区：使用
buffer参数设置适当大小（64KB到1MB）
- 批量操作：减少IO次数，合并写入操作
- 使用mmap：处理大文件时效率更高
- 选择适当格式：二进制格式比文本格式更快
- 避免转换：减少字符串/对象转换
- 并行处理：对多个文件使用并发读取
- 使用索引：需要随机访问时预建索引
- 流式处理：不要一次性加载整个文件到内存
