如何在Python中高效比较多文本文件内容并查找重复行？

我来帮你搞定多文本文件重复行对比的效率问题！你的原始嵌套循环方案慢主要有两个原因：一是用列表做in查找（时间复杂度O(n)），二是重复对比了文件对（比如A和B比完又B和A比），而且循环变量还重复用了i导致逻辑有问题。咱们一步步优化：

第一步：基础效率优化（单进程） #

先从数据结构和对比逻辑入手，用集合代替列表，同时避免重复对比：

import os

def process_files(directory):
    # 筛选出目录下的文件，排除子目录
    files = [os.path.join(directory, f) for f in os.listdir(directory) 
             if os.path.isfile(os.path.join(directory, f))]
    
    # 预读取所有文件的行到集合，自动去重文件内重复行，同时加快查找
    file_line_sets = {}
    for file_path in files:
        with open(file_path, 'r', encoding='utf-8') as f:
            # splitlines()保留每行的原始格式，避免split()拆分单词
            file_line_sets[file_path] = set(f.read().splitlines())
    
    # 只对比i < j的文件对，彻底避免自比和重复对比
    for i in range(len(files)):
        file1 = files[i]
        lines1 = file_line_sets[file1]
        for j in range(i + 1, len(files)):
            file2 = files[j]
            lines2 = file_line_sets[file2]
            # 集合求交集就是重复行，效率极高
            duplicate_lines = lines1 & lines2
            if duplicate_lines:
                print(f"[{file1}] 和 [{file2}] 的重复行:")
                for line in duplicate_lines:
                    print(f"  {line}")

这个版本的优势：

• 集合的in操作和交集计算都是O(1)/O(n)级别的，比嵌套循环快几个数量级
• 用with语句自动管理文件句柄，更安全
• 只做一次i<j的对比，没有冗余计算

第二步：多进程并行优化（大文件/多文件场景） #

如果你的文件数量极多或者单个文件超大，单进程还是会慢，这时候用进程池并行处理IO和对比任务：

import os
from multiprocessing import Pool

def read_file_to_set(file_path):
    """单独封装文件读取函数，供进程池调用"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return file_path, set(f.read().splitlines())
    except Exception as e:
        print(f"读取文件失败 [{file_path}]: {str(e)}")
        return file_path, set()

def compare_file_pair(args):
    """封装文件对比函数，供进程池调用"""
    file1, lines1, file2, lines2 = args
    duplicates = lines1 & lines2
    if duplicates:
        return (file1, file2, duplicates)
    return None

def process_files_with_pool(directory, num_workers=None):
    files = [os.path.join(directory, f) for f in os.listdir(directory) 
             if os.path.isfile(os.path.join(directory, f))]
    
    if len(files) < 2:
        print("至少需要2个文件才能进行对比哦")
        return
    
    # 1. 并行读取所有文件，减少IO等待时间
    with Pool(num_workers) as pool:
        file_data = pool.map(read_file_to_set, files)
    file_line_sets = {fp: lines for fp, lines in file_data}
    
    # 2. 生成所有需要对比的文件对（i<j）
    compare_tasks = []
    for i in range(len(files)):
        file1 = files[i]
        lines1 = file_line_sets[file1]
        for j in range(i + 1, len(files)):
            file2 = files[j]
            lines2 = file_line_sets[file2]
            compare_tasks.append((file1, lines1, file2, lines2))
    
    # 3. 并行对比文件对，利用多核CPU
    with Pool(num_workers) as pool:
        results = pool.map(compare_file_pair, compare_tasks)
    
    # 输出结果
    for res in results:
        if res:
            file1, file2, duplicates = res
            print(f"\n[{file1}] 和 [{file2}] 的重复行:")
            for line in duplicates:
                print(f"  {line}")

这个版本的亮点：

• 把文件读取和对比拆成两个并行阶段，充分利用多核和减少IO阻塞
• 进程池自动管理进程生命周期，不用手动处理进程创建销毁
• 同样严格避免了自比和重复对比

额外注意事项 #

• 如果你的需求是找重复单词而不是行，只需要把splitlines()改成split()（如果要按空格拆分可以用split(' ')，记得过滤空字符串）
• 如果文件大到无法一次性读入内存，可以考虑按块读取并计算哈希值，再对比哈希（这个复杂度高一些，适合超大型文件）
• 编码问题：如果文件有非UTF-8编码，可以用chardet库自动检测编码后再读取

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