如何用Python多线程并行处理500个JSON文件并计算总和？

需求说明 #

你拥有500个JSON文件，每个文件结构示例如下：

{"minute": "2022-11-16T02:29:00.000+00:00", "mycount": [[0, 0], [1, 32], [2, 3456], [3, 446], [4, 534534], [5, 474], [6, 448], [7, 529], [8, 507], [9, 515], [10, 477], [11, 486], [12, 491], [13, 474], [14, 528], [15, 23]]}

需要通过并行处理（同时处理100个文件）完成：

1. 对每个文件，计算mycount中每个子元素第二个值的总和，记为sum1；
2. 计算所有文件的sum1之和，得到总总和。

实现代码 #

使用Python的concurrent.futures.ThreadPoolExecutor可便捷实现多线程并行处理，控制并发数为100，代码如下：

import json
from concurrent.futures import ThreadPoolExecutor
import os

def calculate_sum1(file_path):
    """计算单个JSON文件的sum1值"""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            # 遍历mycount子列表，累加第二个元素
            sum1 = sum(item[1] for item in data['mycount'])
            return sum1
    except Exception as e:
        print(f"处理文件{file_path}出错: {str(e)}")
        return 0  # 出错时返回0，不影响总总和计算

def main():
    # 替换为你的JSON文件所在目录路径
    json_dir = "./json_files"
    # 获取目录下所有JSON文件路径
    file_paths = [os.path.join(json_dir, filename) 
                  for filename in os.listdir(json_dir) 
                  if filename.endswith('.json')]
    
    # 设置并发数为100
    max_workers = 100
    total_sum = 0
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 批量提交任务并获取结果迭代器
        results = executor.map(calculate_sum1, file_paths)
        # 累加所有sum1得到总总和
        total_sum = sum(results)
    
    print(f"所有文件的总总和为: {total_sum}")

if __name__ == "__main__":
    main()

关键说明 #

• 并发控制：通过ThreadPoolExecutor的max_workers参数设置同时处理的文件数为100，匹配需求；
• 单文件处理：calculate_sum1函数负责文件读取、JSON解析与sum1计算，加入异常处理避免单个文件出错导致程序终止；
• 结果收集：executor.map批量提交任务并自动收集返回值，最后通过sum()直接累加得到总总和。

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