如何用多线程/进程优化30万文件网络共享目录的文件名搜索？

先做基础优化（单线程阶段） #

原代码有几个可快速优化的点，能先压缩部分耗时：

• 统一处理oset类型，避免单个字符串传入时的循环冗余
• 若业务允许不区分大小写搜索，提前把关键词和文件名转小写，减少字符串匹配开销
• 用any()替代嵌套循环，逻辑更简洁且执行效率更高

优化后的单线程基础版：

import os

def scan_single_dir(sourcedir: str, oset: set[str]) -> set[str]:
    found = set()
    # 统一转小写（不需要区分大小写则注释掉这两行）
    oset_lower = {ordr.lower() for ordr in oset}
    with os.scandir(sourcedir) as entries:
        for entry in entries:
            if entry.is_file(follow_symlinks=False):
                filename_lower = entry.name.lower()
                if any(ordr in filename_lower for ordr in oset_lower):
                    found.add(entry.name)
    return found

def scan(sourcedir: str, oset: set[str] | str) -> set[str]:
    # 处理输入的关键词类型
    if isinstance(oset, str):
        oset = {oset}
    found = set()
    # 递归遍历所有子目录
    for root, _, _ in os.walk(sourcedir):
        found.update(scan_single_dir(root, oset))
    return found

多线程并行优化方案 #

网络共享目录的扫描属于IO密集型任务，多线程比多进程更适配——进程切换开销大，而线程在等待网络IO时可被调度，能充分利用CPU资源。核心思路是把目录树拆分成多个子目录任务，分配给不同线程并行扫描。

用concurrent.futures.ThreadPoolExecutor实现，每个线程负责扫描一个子目录，最后合并所有结果：

import os
from concurrent.futures import ThreadPoolExecutor, as_completed

def scan_single_dir(sourcedir: str, oset: set[str]) -> set[str]:
    found = set()
    oset_lower = {ordr.lower() for ordr in oset}
    try:
        with os.scandir(sourcedir) as entries:
            for entry in entries:
                if entry.is_file(follow_symlinks=False):
                    filename_lower = entry.name.lower()
                    if any(ordr in filename_lower for ordr in oset_lower):
                        found.add(entry.name)
    except PermissionError:
        # 跳过无权限访问的目录
        pass
    except OSError as e:
        print(f"扫描目录{sourcedir}出错: {e}")
    return found

def scan_parallel(sourcedir: str, oset: set[str] | str, max_workers: int = 16) -> set[str]:
    if isinstance(oset, str):
        oset = {oset}
    
    # 先收集所有需要扫描的子目录（含根目录）
    dirs_to_scan = []
    for root, _, _ in os.walk(sourcedir):
        dirs_to_scan.append(root)
    
    found = set()
    # 用线程池并行扫描
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有目录扫描任务
        future_to_dir = {executor.submit(scan_single_dir, d, oset): d for d in dirs_to_scan}
        # 逐个获取任务结果并合并
        for future in as_completed(future_to_dir):
            found.update(future.result())
    
    return found

# 调用示例
if __name__ == "__main__":
    target_dir = r"\\your-network-share\path"
    keywords = {"order", "invoice"}
    result = scan_parallel(target_dir, keywords)
    for name in result:
        print(name)

关键说明 #

1. 线程数设置：max_workers建议设为16-32（网络IO密集型任务，线程数可多于CPU核心数，因为多数时间线程在等待网络响应），可根据实际网络带宽调整。
2. 异常处理：添加权限错误和IO错误捕获，避免单个目录扫描失败导致整个任务终止。
3. 动态文件增删：因目录动态变化，扫描过程中可能出现文件新增/删除导致的结果偏差，这是分布式文件系统的固有特性，若需更精确结果，可考虑定期增量扫描。
4. 性能提升：原单线程耗时395秒，并行扫描后通常能将耗时压缩至原有的1/5~1/10（具体取决于网络带宽和目录结构）。

为什么不用多进程？ #

多进程适配CPU密集型任务，而网络目录扫描的瓶颈是网络IO等待，多进程的进程间通信和切换开销会抵消并行收益，反而不如多线程高效。

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