使用Python多进程批量处理数千JSON文件遇报错的解决方案咨询

错误原因 #

你一次性创建了与JSON文件数量一致的进程（数千个），每个进程都会打开一个文件，而操作系统对单个用户能同时打开的文件描述符数量有默认限制（通常在几百到几千之间），当并发打开的文件数超过这个限制时，就会触发OSError: [Errno 24] Too many open files。

解决方案 #

1. 使用进程池控制并发数（推荐） #

用multiprocessing.Pool替代手动创建大量进程，进程池会自动限制并发进程的数量（建议设置为CPU核心数或略高，比如os.cpu_count()），既保证处理效率，又不会耗尽系统资源。

修改后的完整代码：

import json
import os
from multiprocessing import Pool

def read_file(file_name):
    '''读取指定JSON文件并返回数据'''
    try:
        with open(file_name) as file:
            return json.load(file)
    except Exception as e:
        print(f"读取文件{file_name}失败: {str(e)}")
        return None

def do_some_process(data):
    '''这里编写你的数据处理逻辑'''
    if data is None:
        return None
    # 示例处理逻辑，根据实际需求替换
    some_result = {"processed_length": len(data)}
    return some_result

def process_each_file(file):
    '''单个文件的处理流程，返回（文件名前缀，处理结果）'''
    file_name = file.split('.')[0]
    data = read_file(os.path.join("../data", file))
    processed_result = do_some_process(data)
    return (file_name, processed_result)

if __name__ == '__main__':
    # 筛选出目录下的所有JSON文件
    file_list = [f for f in os.listdir("../data") if f.endswith(".json")]
    
    # 创建进程池，并发数设为CPU核心数
    with Pool(processes=os.cpu_count()) as pool:
        # 批量执行处理任务
        results = pool.map(process_each_file, file_list)
    
    # 将结果转换为字典，过滤掉处理失败的项
    result_dict = {k: v for k, v in results if v is not None}
    
    # 后续处理逻辑，根据需求修改
    print(f"成功处理{len(result_dict)}个文件")

2. 优化文件读取的异常处理 #

在read_file函数中加入异常捕获，避免单个文件读取失败导致进程崩溃，同时确保文件描述符被正确释放，进一步降低资源泄漏的风险。

3. 临时调整系统文件描述符限制（可选） #

如果必须维持高并发，可临时提高系统的文件描述符上限（仅当前终端会话生效，重启后恢复）：

• Linux/macOS终端执行：ulimit -n 4096
• 但这只是临时 workaround，不如进程池方案从根源解决问题。

补充说明 #

使用进程池的核心优势：

• 自动复用进程，减少进程创建/销毁的额外开销
• 严格控制并发数，避免资源耗尽
• 内置的map/imap方法简化批量任务的调度逻辑

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