如何实现多目录压缩文件并行解压 提升现有批量处理代码运行效率

你的场景属于IO+CPU混合密集型任务，13个目录的处理逻辑完全独立无依赖，非常适合用多进程并行改造，改造后总耗时可压缩到单目录处理的7分钟左右，符合你的预期。

改造步骤 #

• 把原有单个zip文件的处理逻辑封装为独立的任务函数，避免进程间变量共享的额外开销
• 先扫描全量符合匹配规则的zip文件，生成任务参数列表
• 调用多进程池批量执行所有任务

完整改造后代码 #

import os
import fnmatch
import re
import datetime
import zipfile
from multiprocessing import Pool

# 处理单个zip文件的独立函数，所有依赖参数通过入参传入
def process_single_zip(params):
    zip_path, data_files, counter_part = params
    filename = os.path.basename(zip_path)
    
    # 原业务逻辑保留，仅做资源释放优化
    date_zipped_file_s = re.search('-(.\d+)-', filename).group(1)
    date_zipped_file = datetime.datetime.strptime(date_zipped_file_s, '%Y%m%d').date()      
          
    # 计算目标路径
    new_dir = os.path.normpath(os.path.join(os.path.relpath(zip_path, start=data_files), ".."))
    new = os.path.join(counter_part, new_dir)

    # 创建目录，exist_ok=True避免重复判断和并发创建冲突
    os.makedirs(new, exist_ok=True)
    
    # 用上下文管理器打开zip文件，自动释放资源
    with zipfile.ZipFile(zip_path) as zf:
        zf.extractall(new) 

    # 重命名逻辑保留
    files = os.listdir(new)
    for file in files:
        filesplit = os.path.splitext(os.path.basename(file))
        if not re.search(r'_\d{8}.', file):
            os.rename(os.path.join(new, file), os.path.join(new, filesplit[0]+'_'+date_zipped_file_s+filesplit[1]))

if __name__ == '__main__':
    # 这里替换为你自己的实际路径
    data_files = "/your/source/root/path"
    counter_part = "/your/target/root/path"
    pattern = '*.zip'
    
    # 第一步：先扫描所有符合条件的zip文件，生成任务列表
    task_list = []
    for root, dirs, files in os.walk(data_files): 
        for filename in fnmatch.filter(files, pattern):
            zip_path = os.path.join(root, filename)
            task_list.append( (zip_path, data_files, counter_part) )
    
    # 第二步：初始化多进程池，processes可根据实际情况调整，13个任务直接设13即可
    # 如果CPU核心数较少，也可以设为os.cpu_count()
    with Pool(processes=13) as pool:
        pool.map(process_single_zip, task_list)

注意事项 #

• 若使用机械硬盘，并行数不要超过20，避免磁盘IO抢占导致效率下降，13个任务适配绝大多数存储设备
• 路径建议统一使用绝对路径，避免多进程工作目录不一致引发的路径错误
• 原有业务逻辑完全保留，仅优化了资源管理和并发调度，不会影响原有处理结果

内容的提问来源于stack exchange，提问作者Mediterráneo