如何用Python多线程加速三级嵌套字典的分片存储任务？

核心思路 #

把嵌套字典的所有文件写入任务拆解为扁平的任务元组列表，每个元组包含(文件名, 待写入内容)，再用线程池并行执行所有写入操作——文件IO属于IO密集型任务，多线程能有效利用等待IO的时间提升整体效率。

代码实现 #

1. 定义文件写入函数 #

这是线程执行的最小单元，负责单个文件的写入逻辑：

import json
from concurrent.futures import ThreadPoolExecutor

def write_to_file(filename, content):
    # 示例用JSON序列化存储，可根据需求替换为文本、CSV等其他格式
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(content, f, ensure_ascii=False, indent=2)

2. 生成所有任务列表 #

遍历嵌套字典，把每个层级的写入任务整理成扁平列表：

def generate_tasks(nested_dict):
    tasks = []
    for k1, level2_data in nested_dict.items():
        # 一级文件任务
        tasks.append((f"{k1}.txt", level2_data))
        for k2, level3_data in level2_data.items():
            # 二级文件任务
            tasks.append((f"{k1}_{k2}.txt", level3_data))
            for k3, final_value in level3_data.items():
                # 三级文件任务
                tasks.append((f"{k1}_{k2}_{k3}.txt", final_value))
    return tasks

3. 多线程执行任务 #

用ThreadPoolExecutor批量调度所有任务：

if __name__ == "__main__":
    # 替换为你的实际三级嵌套字典
    your_large_dict = {
        "user_a": {
            "order_001": {
                "item_1": {"name": "apple", "count": 5},
                "item_2": {"name": "banana", "count": 3}
            },
            "order_002": {
                "item_3": {"name": "orange", "count": 2}
            }
        },
        # 更多数据...
    }

    # 生成所有待执行任务
    task_list = generate_tasks(your_large_dict)

    # 线程池大小建议：IO密集型任务可设为CPU核心数的2-4倍，或根据实际测试调整
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 用lambda拆包任务元组，传递给写入函数
        executor.map(lambda task_args: write_to_file(*task_args), task_list)

关键说明 #

• 任务独立性：每个文件写入任务完全独立，按你的命名规则不会出现多任务写入同一文件的情况，因此无需额外加锁。
• 线程池大小调整：如果存储系统IO性能较好，可适当调大max_workers（比如20）；若IO性能一般，过大的线程数反而会导致IO阻塞，建议从小值测试优化。
• 内容格式适配：如果字典值是JSON不可序列化类型（如自定义对象），需在write_to_file中提前做格式转换（比如转为字符串、自定义序列化逻辑）。
• 文件名安全：如果k1/k2/k3包含特殊字符（如斜杠、空格），建议提前做字符替换（比如把/换成_），避免生成无效文件名。

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