速度优先的哈希计算与过滤实现方案及Python脚本优化技巧

原脚本的核心性能瓶颈集中在频繁的文件开关、冗余的字符串扫描、低效率的IO操作三个方面，可通过以下技巧优化：

• 优化1：缓存目标文件句柄，避免每行打开一次文件
  原脚本每处理一行就执行一次open(...).write(...)，执行完立即关闭文件，单次系统调用开销是单行写入的几十上百倍。可以用字典缓存所有已打开的目标文件句柄，MD5前4位最多只有65536种可能，所有句柄完全可以存放在内存中，所有内容写完后统一关闭即可。
• 优化2：简化文件遍历逻辑，避免手动路径拼接
  原脚本用双层os.listdir加字符串拼接路径，容易出错且效率低，改用pathlib.Path.glob递归遍历所有文件，代码更简洁，路径处理也更安全高效。
• 优化3：优化字符串分割逻辑，减少重复扫描
  原脚本逐行扫描3次判断分隔符，改用预编译的正则表达式一次性分割，只扫描一次字符串，还可以指定最多分割1次，避免后半段内容里的分隔符被错误切割。
• 优化4：批量缓存写入内容，减少IO次数
  可以给每个目标文件设置一个写入缓存，比如攒够1000行再一次性写入磁盘，进一步降低IO操作的频次。
• 优化5：使用上下文管理器管理文件句柄，避免泄漏
  原脚本打开的读文件没有主动关闭，长时间运行会占用过多系统句柄，改用with关键字打开文件，自动处理关闭逻辑。
• 优化6：多线程并行处理（适配IO密集场景）
  该任务属于典型的IO密集型，大部分时间消耗在磁盘读写上，用concurrent.futures.ThreadPoolExecutor并行处理多个输入文件，CPU等待IO的间隙可以处理其他文件，大幅提升整体吞吐量。
• 优化7：修复变量名冲突问题
  原脚本用hash作为变量名，覆盖了Python内置的hash()函数，存在潜在风险，修改为其他名称即可。

优化后参考代码 #

import os
import re
import hashlib
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

# 配置参数
INPUT_ROOT = Path("Folder")
OUTPUT_DIR = Path("HashDB")
CACHE_LIMIT = 1000  # 每个文件攒满多少行再写入

# 初始化输出目录
OUTPUT_DIR.mkdir(exist_ok=True)
# 缓存文件句柄和写入缓存
file_handles = {}
write_caches = {}
# 预编译正则，避免每次运行都编译
split_pattern = re.compile(r'[:; ]', maxsplit=1)

def process_single_file(file_path: Path):
    with open(file_path, 'r', encoding="ISO-8859-1") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            # 单次分割，取后半段
            split_res = split_pattern.split(line, maxsplit=1)
            if len(split_res) < 2:
                continue
            final_line = split_res[1]
            # 计算md5
            md5_val = hashlib.md5(final_line.encode('utf-8')).hexdigest()
            target_prefix = md5_val[:4]
            target_line = f"{md5_val}:{final_line}\n"
            # 写入缓存
            if target_prefix not in write_caches:
                write_caches[target_prefix] = []
            write_caches[target_prefix].append(target_line)
            # 达到缓存上限则写入
            if len(write_caches[target_prefix]) >= CACHE_LIMIT:
                if target_prefix not in file_handles:
                    file_handles[target_prefix] = open(OUTPUT_DIR / f"{target_prefix}.txt", 'a', encoding="ISO-8859-1")
                file_handles[target_prefix].writelines(write_caches[target_prefix])
                write_caches[target_prefix].clear()

if __name__ == "__main__":
    # 递归获取所有输入文件
    all_files = [f for f in INPUT_ROOT.rglob("*") if f.is_file()]
    # 多线程处理，线程数可按CPU核数*2调整
    with ThreadPoolExecutor(max_workers=os.cpu_count()*2) as pool:
        pool.map(process_single_file, all_files)
    # 所有文件处理完后，写入剩余缓存，关闭所有句柄
    for prefix, cache in write_caches.items():
        if cache:
            if prefix not in file_handles:
                file_handles[prefix] = open(OUTPUT_DIR / f"{prefix}.txt", 'a', encoding="ISO-8859-1")
            file_handles[prefix].writelines(cache)
    for fh in file_handles.values():
        fh.close()

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