如何利用多进程实现两个超大文本列表的差异查找（解决内存不足与效率低下问题）

哇，处理几十GB级别的文本列表确实是个头疼的问题，你的pandas嵌套分块写法效率低是必然的——想想看，你每处理列表1的一行，就要把整个76GB的列表2从头读到尾，这复杂度直接是O(n*m)，慢到离谱也正常。而且多进程如果没找对方向，确实也搞不定。我给你几个靠谱的解决方案，从最快的系统命令到Python实现都有：

方案一：用系统命令（最快最省心，优先推荐） #

如果你的环境是Linux/macOS，直接用系统自带的sort和comm命令就搞定了，这些工具是C语言实现的，经过高度优化，处理大文件的效率比Python高几个数量级，完全不用担心内存问题：

1. 先给两个文件排序（sort会自动处理大文件的分块排序，不用操心内存）：

   sort /folder/list1 -o list1_sorted.txt
   sort /folder/list2 -o list2_sorted.txt
   

2. 用comm命令提取列表1独有的行：

   comm -23 list1_sorted.txt list2_sorted.txt > result.txt
   

   解释一下参数：
   • -2：去掉列表2独有的行
   • -3：去掉两个列表共有的行
   • 剩下的就是列表1里有、列表2里没有的行，完美符合你的需求。

方案二：Python多进程+哈希块预处理（适合不能用系统命令的场景） #

如果必须用Python实现，核心思路是先把列表2预处理成可快速查找的哈希集合分块，然后用多进程并行处理列表1的分块，避免重复读取整个列表2：

第一步：预处理列表2，生成哈希块 #

把列表2分成多个小的哈希集合，存在磁盘上（这样不用一次性把76GB的内容全塞进内存）：

import pickle
from pathlib import Path

def split_list2_into_hash_chunks(input_path, chunk_size=1_000_000, output_dir="list2_chunks"):
    # 创建存储分块的目录
    Path(output_dir).mkdir(exist_ok=True)
    chunk_idx = 0
    current_set = set()
    
    with open(input_path, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f):
            cleaned_line = line.strip()
            if cleaned_line:  # 跳过空行
                current_set.add(cleaned_line)
            # 达到块大小就保存
            if (line_num + 1) % chunk_size == 0:
                with open(f"{output_dir}/chunk_{chunk_idx}.pkl", 'wb') as pkl_file:
                    pickle.dump(current_set, pkl_file)
                current_set = set()
                chunk_idx += 1
    # 处理最后剩余的元素
    if current_set:
        with open(f"{output_dir}/chunk_{chunk_idx}.pkl", 'wb') as pkl_file:
            pickle.dump(current_set, pkl_file)

# 运行预处理
split_list2_into_hash_chunks("/folder/list2")

第二步：多进程处理列表1 #

加载预处理好的哈希块，用多进程并行检查列表1的每一行是否存在于列表2：

import pickle
from pathlib import Path
from multiprocessing import Pool, cpu_count

# 加载所有哈希块到内存（如果内存不够，可以改成每次查询时读取，但效率会稍降）
list2_chunks = []
chunk_dir = Path("list2_chunks")
for chunk_file in chunk_dir.glob("chunk_*.pkl"):
    with open(chunk_file, 'rb') as f:
        list2_chunks.append(pickle.load(f))

def is_line_in_list2(line):
    cleaned_line = line.strip()
    if not cleaned_line:
        return False
    # 遍历所有哈希块检查
    for chunk_set in list2_chunks:
        if cleaned_line in chunk_set:
            return True
    return False

def process_list1_chunk(chunk_lines):
    # 处理一个列表1的分块，返回不在列表2的行
    return [line for line in chunk_lines if not is_line_in_list2(line)]

def main():
    input_path = "/folder/list1"
    output_path = "result.txt"
    chunk_size = 1_000_000  # 可根据你的内存情况调整
    
    # 用CPU核心数创建进程池
    with Pool(cpu_count()) as pool:
        with open(input_path, 'r', encoding='utf-8') as f, open(output_path, 'w', encoding='utf-8') as out_f:
            current_chunk = []
            for line in f:
                current_chunk.append(line.strip())
                # 达到块大小就提交给进程池处理
                if len(current_chunk) == chunk_size:
                    results = pool.apply(process_list1_chunk, (current_chunk,))
                    for res_line in results:
                        out_f.write(res_line + '\n')
                    current_chunk = []
            # 处理最后一个剩余的块
            if current_chunk:
                results = pool.apply(process_list1_chunk, (current_chunk,))
                for res_line in results:
                    out_f.write(res_line + '\n')

if __name__ == "__main__":
    main()

为什么你的原代码效率这么低？ #

你的代码最大的问题是嵌套循环读取列表2：每处理列表1的一行，就要完整遍历一遍列表2的所有分块。假设列表1有1亿行，那就要读1亿次76GB的列表2，这完全是灾难级的IO开销，效率不可能高。而上面的方案都是把列表2只读取一次，预处理成可快速查找的结构，之后只需要做O(1)的哈希查询，效率提升几个数量级。

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