如何加速Python中处理大文件字符串的re.sub()操作？

原代码的核心性能瓶颈有两个：

1. 全量加载超大文件：20万+行的文件一次性读入内存，既占用大量内存，也让正则处理超大字符串时效率极低。
2. 正则回溯开销过高：((?:^.*\n)+)(?=\1) 这种模式会让正则引擎尝试所有可能的多行序列，再通过正向预查验证重复，对于大文本来说，这种回溯的时间复杂度是指数级的，直接导致处理耗时长达10分钟以上。

优化思路 #

1. 逐行/分块处理，避免全量加载 #

不再一次性读取整个文件，用迭代器或缓冲区逐行读取，降低内存占用的同时，避免正则处理超大字符串的低效。

2. 手动跟踪重复块，替代正则匹配 #

通过维护当前块缓冲区，逐行对比后续内容是否与当前块重复，一旦发现重复就跳过所有重复部分，这种线性扫描的时间复杂度为O(n)，远快于正则的回溯逻辑。

3. 可选：哈希加速重复判断 #

对于较长的重复块，先计算块的哈希值，通过对比哈希快速判断是否重复，避免逐行比较的开销。

实现代码 #

基础版：逐行匹配重复块 #

from collections import deque

def deduplicate_consecutive_blocks(input_path, output_path):
    with open(input_path, 'r') as infile, open(output_path, 'w') as outfile:
        # 用队列存储预读取的行，避免重复读取文件
        line_queue = deque()
        current_block = []
        
        def get_next_line():
            """从队列或文件中获取下一行"""
            if line_queue:
                return line_queue.popleft()
            return infile.readline()
        
        while True:
            line = get_next_line()
            if not line:
                # 文件读取完毕，写入剩余的块
                if current_block:
                    outfile.writelines(current_block)
                break
            
            current_block.append(line)
            block_length = len(current_block)
            match = True
            # 检查后续是否有和当前块完全一致的内容
            for idx in range(block_length):
                candidate_line = get_next_line()
                if not candidate_line or candidate_line != current_block[idx]:
                    match = False
                    # 把不匹配的行放回队列
                    if candidate_line:
                        line_queue.appendleft(candidate_line)
                    break
                # 匹配的行先暂存，后续确认重复后丢弃
                line_queue.append(candidate_line)
            
            if match:
                # 找到重复块，跳过所有后续的重复部分
                while True:
                    next_match = True
                    temp_lines = []
                    for bline in current_block:
                        cline = get_next_line()
                        if not cline or cline != bline:
                            next_match = False
                            if cline:
                                temp_lines.append(cline)
                            break
                        temp_lines.append(cline)
                    if not next_match:
                        # 把不匹配的行放回队列
                        while temp_lines:
                            line_queue.appendleft(temp_lines.pop())
                        break
                # 写入当前块，重置缓冲区
                outfile.writelines(current_block)
                current_block = []
            else:
                # 当前块无重复，若块过长则直接写入，避免无限扩展
                if len(current_block) > 200:
                    outfile.writelines(current_block)
                    current_block = []

# 使用示例
deduplicate_consecutive_blocks('largefile.txt', 'deduplicated.txt')

优化版：哈希加速重复判断 #

对于较长的重复块，哈希对比可以大幅减少逐行比较的时间：

import hashlib
from collections import deque

def calculate_block_hash(block):
    """计算块的MD5哈希，用于快速判断重复"""
    block_content = ''.join(block).encode('utf-8')
    return hashlib.md5(block_content).hexdigest()

def deduplicate_consecutive_blocks(input_path, output_path):
    with open(input_path, 'r') as infile, open(output_path, 'w') as outfile:
        line_queue = deque()
        current_block = []
        # 预读取一批行到缓冲区
        for _ in range(1000):
            line = infile.readline()
            if not line:
                break
            line_queue.append(line)
        
        while current_block or line_queue:
            if not current_block:
                if not line_queue:
                    break
                current_block.append(line_queue.popleft())
            
            block_len = len(current_block)
            # 缓冲区有足够行，尝试判断重复
            if len(line_queue) >= block_len:
                candidate_block = line_queue[:block_len]
                if calculate_block_hash(current_block) == calculate_block_hash(candidate_block):
                    # 哈希匹配，再逐行验证避免哈希碰撞
                    if current_block == candidate_block:
                        # 跳过缓冲区中的重复块
                        del line_queue[:block_len]
                        # 继续读取文件，跳过所有后续重复块
                        while True:
                            next_candidate = []
                            for _ in range(block_len):
                                line = infile.readline()
                                if not line:
                                    break
                                next_candidate.append(line)
                            if len(next_candidate) != block_len or current_block != next_candidate:
                                line_queue = next_candidate + line_queue
                                break
                        # 写入当前块，重置
                        outfile.writelines(current_block)
                        current_block = []
                    else:
                        # 哈希碰撞，扩展当前块
                        current_block.append(line_queue.popleft())
                else:
                    # 不匹配，扩展当前块
                    current_block.append(line_queue.popleft())
                # 块过长直接写入
                if len(current_block) > 200:
                    outfile.writelines(current_block)
                    current_block = []
            else:
                # 缓冲区不足，读取更多行
                for _ in range(1000):
                    line = infile.readline()
                    if not line:
                        break
                    line_queue.append(line)
                if len(line_queue) < block_len:
                    # 剩余行不足，写入当前块
                    outfile.writelines(current_block)
                    current_block = []
        # 写入剩余缓冲区内容
        if line_queue:
            outfile.writelines(line_queue)

额外建议 #

• 如果需要覆盖原文件，可以在处理完成后用shutil.move('deduplicated.txt', 'largefile.txt')替换。
• 调整current_block的最大长度（比如200）可以平衡性能和内存占用，根据实际文件的重复块长度调整。

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