如何使用Python以并行方式遍历JSON树（字典）？

问题场景 #

给定如下结构的JSON数据：

{ 
  'A':
     {
       'a':[],
       'aa': 'val'
     },
  'B':
     { 
       'b':[],
       'bb': 'val'
     },
  'C':
     {
       'c':[],
       'cc': 'val'
     }
}

该JSON包含上千个类似A、B、C的顶层分支，逐个串行遍历搜索目标值val效率极低，需要通过并行方式同时在所有分支中遍历搜索。

解决方案 #

可以使用Python的concurrent.futures模块实现并行搜索，以下是具体实现：

1. 单分支搜索函数

先定义一个递归遍历单分支的搜索函数，负责检查分支内是否存在目标值：

def search_branch(branch_data, target_val):
    def recursive_search(data):
        if isinstance(data, dict):
            for _, v in data.items():
                if v == target_val:
                    return True
                if recursive_search(v):
                    return True
        elif isinstance(data, list):
            for item in data:
                if recursive_search(item):
                    return True
        return False
    return recursive_search(branch_data)

2. 并行搜索主逻辑

根据场景选择线程池（IO密集型优先）或进程池（CPU密集型优先）并行处理所有分支：

import concurrent.futures

def parallel_search(json_data, target_val):
    results = {}
    # 线程池适合IO密集型任务，若为CPU密集型可替换为ProcessPoolExecutor
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
        # 提交所有分支的搜索任务
        task_map = {executor.submit(search_branch, data, target_val): branch for branch, data in json_data.items()}
        # 逐个获取任务结果
        for future in concurrent.futures.as_completed(task_map):
            branch_name = task_map[future]
            try:
                results[branch_name] = future.result()
            except Exception as exc:
                results[branch_name] = f"搜索异常: {exc}"
    return results

# 示例调用
if __name__ == "__main__":
    test_json = { 
        'A': {'a':[], 'aa': 'val'},
        'B': {'b':[], 'bb': 'not_val'},
        'C': {'c':[], 'cc': 'val'}
    }
    target = 'val'
    print(parallel_search(test_json, target))
    # 输出: {'A': True, 'C': True, 'B': False}

注意事项

• 若JSON分支的遍历计算量极大，改用ProcessPoolExecutor规避GIL限制，max_workers建议设为CPU核心数。
• 线程池的max_workers可根据机器性能调整，通常设为CPU核心数的2倍即可。
• 异常处理确保单个分支搜索出错不会导致整个并行任务崩溃。

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