处理大文件是Python开发中的常见挑战，不当的处理方式会导致内存不足或性能问题。以下是经过实践验证的最佳方案。

一、核心原则

1. 绝不一次性读取大文件：避免read()或readlines()方法
2. 优先使用迭代：最安全的内存友好方式
3. 随机访问需求用mmap：当需要随机访问时选择内存映射
4. 考虑I/O性能：合理设置缓冲区大小

二、迭代处理最佳实践

1. 文本文件逐行处理（推荐方案）

def process_large_text(file_path):
    """处理超大文本文件的标准方法"""
    with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
        line_count = 0
        for line in f:  # 内存高效迭代
            line_count += 1
            # 处理每行内容
            processed = line.strip().upper()
            yield processed  # 使用生成器进一步节省内存
            
        print(f"总共处理了 {line_count} 行")

2. 二进制文件分块处理

def chunked_file_reader(file_path, chunk_size=1024*1024):
    """二进制文件分块读取器"""
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 使用示例
for chunk in chunked_file_reader('large_video.mp4'):
    analyze_chunk(chunk)

3. 带重叠的分块处理（防止边界问题）

def sliding_chunk_reader(file_path, chunk_size=1_000_000, overlap=100):
    """带重叠区的分块读取，防止跨块数据截断"""
    with open(file_path, 'rb') as f:
        prev_chunk = b''
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            
            # 拼接前块的末尾作为重叠区
            full_chunk = prev_chunk[-overlap:] + chunk if prev_chunk else chunk
            yield full_chunk
            
            # 保留当前块的末尾作为下一个块的重叠区
            prev_chunk = chunk
            
            # 回退重叠区大小的位置
            if len(chunk) == chunk_size:
                f.seek(-overlap, 1)

三、内存映射高级技巧

1. 安全的内存映射实现

import mmap
import contextlib

def safe_mmap(file_path, access=mmap.ACCESS_READ):
    """带错误处理的内存映射"""
    try:
        with open(file_path, 'r+b') as f:
            with contextlib.closing(mmap.mmap(f.fileno(), 0, access=access)) as mm:
                yield mm
    except (ValueError, PermissionError) as e:
        print(f"内存映射失败: {e}")
        raise

2. 高效搜索实现

def mmap_search(file_path, pattern):
    """使用mmap实现高效搜索"""
    pattern = pattern.encode('utf-8') if isinstance(pattern, str) else pattern
    
    with safe_mmap(file_path) as mm:
        pos = mm.find(pattern)
        while pos != -1:
            yield pos
            pos = mm.find(pattern, pos + 1)

3. 大文件编辑技巧

def mmap_replace(file_path, old, new):
    """内存映射实现高效替换"""
    old = old.encode('utf-8') if isinstance(old, str) else old
    new = new.encode('utf-8') if isinstance(new, str) else new
    
    assert len(old) == len(new), "替换内容长度必须相同"
    
    with safe_mmap(file_path, access=mmap.ACCESS_WRITE) as mm:
        pos = mm.find(old)
        while pos != -1:
            mm[pos:pos+len(old)] = new
            pos = mm.find(old, pos + len(new))

四、性能优化方案

1. 缓冲区大小调优

# 根据文件大小自动调整缓冲区
def optimized_buffer_size(file_size):
    """智能缓冲区大小计算"""
    if file_size < 10*1024*1024:    # <10MB
        return 64*1024              # 64KB
    elif file_size < 1*1024*1024*1024:  # <1GB
        return 1*1024*1024          # 1MB
    else:
        return 10*1024*1024         # 10MB

with open('large.file', 'rb', buffering=optimized_buffer_size(os.path.getsize('large.file'))) as f:
    for chunk in iter(lambda: f.read(optimized_buffer_size(f.size)), b''):
        process(chunk)

2. 多核并行处理

from multiprocessing import Pool

def parallel_file_processor(file_path, workers=4):
    """多进程并行处理大文件"""
    def worker(args):
        offset, size = args
        with open(file_path, 'rb') as f:
            f.seek(offset)
            chunk = f.read(size)
            return process_chunk(chunk)
    
    file_size = os.path.getsize(file_path)
    chunk_size = file_size // workers
    
    # 计算各worker的任务范围
    tasks = []
    for i in range(workers):
        start = i * chunk_size
        end = start + chunk_size if i != workers-1 else file_size
        tasks.append((start, end - start))
    
    with Pool(workers) as pool:
        results = pool.map(worker, tasks)
    
    return combine_results(results)

五、异常处理与健壮性

1. 全面的错误处理

def robust_file_processor(file_path):
    try:
        file_size = os.path.getsize(file_path)
        if file_size > 100*1024*1024:  # >100MB
            print("警告：处理大文件，可能需要较长时间")
            
        with open(file_path, 'rb') as f:
            # 根据文件大小选择处理策略
            if file_size > 1*1024*1024*1024:  # >1GB
                processor = chunked_file_reader(f)
            else:
                processor = f
                
            for data in processor:
                try:
                    process(data)
                except ProcessingError as e:
                    log_error(f"数据处理失败: {e}")
                    continue
                    
    except FileNotFoundError:
        print(f"文件不存在: {file_path}")
    except PermissionError:
        print(f"无访问权限: {file_path}")
    except IOError as e:
        print(f"I/O错误: {e}")
    except Exception as e:
        print(f"未知错误: {e}")

六、场景化解决方案

1. 超大CSV文件处理

import pandas as pd

def process_large_csv(csv_path):
    """分块读取超大CSV文件"""
    chunk_size = 10**6  # 每次读取1百万行
    for chunk in pd.read_csv(csv_path, chunksize=chunk_size):
        process_dataframe(chunk)

2. 日志文件实时监控

def tail_large_log(file_path):
    """模拟tail -f功能，实时读取日志新增内容"""
    with open(file_path, 'r', encoding='utf-8') as f:
        # 先定位到文件末尾
        f.seek(0, 2)
        
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.1)  # 短暂休眠
                continue
            yield line

3. 二进制文件模式匹配

def binary_pattern_search(file_path, pattern, chunk_size=1024*1024):
    """在二进制文件中搜索模式(跨块安全)"""
    pattern = re.compile(pattern) if isinstance(pattern, str) else pattern
    buffer = b''
    
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
                
            buffer += chunk
            matches = list(pattern.finditer(buffer))
            
            for match in matches[:-1]:  # 处理所有完整匹配
                yield match.group()
                
            # 保留最后不完整的部分
            buffer = buffer[matches[-1].start():] if matches else buffer[-len(pattern):]

七、总结对比表

终极建议：

1. 文本文件优先使用逐行迭代
2. 二进制文件使用分块处理
3. 需要随机访问时选择内存映射
4. TB级文件考虑并行处理方案

通过合理应用这些技术，可以高效安全地处理从GB到TB级别的各种文件，同时保持代码的健壮性和可维护性。

Tags：python实践

微信扫一扫：分享

微信里点“发现”，扫一下

二维码便可将本文分享至朋友圈。