Python IO优化案例如何优化文件读写

wen python案例 2026-06-28 286

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()