Python大文件分块合并性能优化：多线程读取效率提升咨询

核心问题分析 #

你的性能瓶颈本质是随机行块读取的磁盘IO开销：Python默认按行定位指定行时，必须从头逐行扫描文件，对20GB级大文件来说这是O(n)的低效操作。ThreadPoolExecutor不仅无法解决磁盘IO的本质瓶颈，反而可能因线程切换、磁盘寻道竞争进一步拖慢速度。

现有代码的针对性优化 #

1. 预生成行偏移索引（最关键优化） #

提前遍历一次文件A、B，记录每一行的字节偏移量，后续读取指定行块时直接通过seek()跳转，彻底避免重复扫描：

def build_line_index(file_path):
    index = []
    with open(file_path, 'rb') as f:
        offset = 0
        index.append(offset)
        while f.readline():
            offset = f.tell()
            index.append(offset)
    return index

# 预先生成索引（只需执行一次）
index_a = build_line_index('fileA.txt')
index_b = build_line_index('fileB.txt')

# 读取指定行块（左闭右开区间：start_line到end_line）
def read_block(file_path, index, start_line, end_line):
    with open(file_path, 'rb') as f:
        f.seek(index[start_line])
        # 一次性读取整个行块的字节，减少IO次数
        byte_count = index[end_line] - index[start_line]
        return f.read(byte_count).decode('utf-8')  # 根据实际编码调整

2. 替换线程池为单线程顺序执行 #

磁盘是顺序设备，随机IO场景下多线程会导致频繁寻道，单线程顺序处理反而能利用磁盘预读机制，大幅减少寻道开销。若必须并行，优先用multiprocessing.Pool（需注意进程间文件句柄共享问题），但多数场景下单线程足够。

3. 增大IO缓冲区 #

打开文件时设置较大的缓冲区，减少系统IO调用次数：

with open(file_path, 'rb', buffering=1024*1024*10) as f:  # 10MB缓冲区
    pass

4. 批量读取而非逐行处理 #

利用read()直接读取整个行块的字节内容，避免逐行readline()的多次系统调用，如上述read_block函数所示。

更优的大文件分块合并策略 #

1. 基于内存映射（mmap）的随机访问 #

使用mmap将大文件映射到内存，把磁盘IO转化为内存访问，大幅提升随机读取速度：

import mmap

def read_block_mmap(file_path, index, start_line, end_line):
    with open(file_path, 'rb') as f:
        with mmap.mmap(f.fileno(), length=0, access=mmap.ACCESS_READ) as mm:
            start = index[start_line]
            end = index[end_line]
            return mm[start:end].decode('utf-8')

2. 预规划合并顺序，减少磁盘寻道 #

若业务允许，尽量将同一文件的连续行块合并读取，再写入目标文件，而非严格交替读取A、B的零散行块。例如：先读取A的所有连续行块缓存（内存足够时），再读取B的对应行块，交替写入C，减少磁盘来回寻道的时间。

3. 跳过行边界检测（若业务允许） #

如果对行块的精度要求不高，可以直接按字节块划分（比如每块100MB），跳过行边界检测，直接读取字节块并写入，这是最快的合并方式，但需确保业务能接受少量行的跨块情况。

4. 调用系统级工具辅助 #

对于Linux/macOS，可利用dd、cat等系统工具实现分块读取，Python通过subprocess调用，系统级工具的IO效率远高于Python原生代码：

import subprocess

def write_block_to_c(file_path, start_offset, byte_count):
    with open('fileC.txt', 'ab') as c:
        subprocess.run(
            ['dd', f'if={file_path}', f'skip_bytes={start_offset}', f'count_bytes={byte_count}'],
            stdout=c,
            check=True
        )

内容的提问来源于stack exchange，提问作者lore10