无文件IO场景下Python multiprocess/multiprocessing报OSError Errno24打开文件过多

• 需求：使用Python multiprocess 和 multiprocessing 库实现并行处理，需满足3项要求：
  1. 使用共享数据结构
  2. 向目标函数传递多参数
  3. 进程数量根据当前系统配置自适应调整
• 故障表现：小任务量运行正常，任务量增大后触发错误 OSError: [Errno 24] Too many open files
• 运行环境：macOS Catalina，PyCharm内执行 ulimit -n 得到系统打开文件数限制为1024
• 约束：不修改 ulimit 配置，代码需适配多系统、开箱即用
• 已尝试方案：使用.join()、gc.collect()、关闭非必要文件（代码未操作业务文件），均无效

复现代码 #

import gc
import time

import numpy as np

from math import pi
from multiprocess import Process, Manager
from multiprocessing import Semaphore, cpu_count

def do_work(element, shared_array, sema):
    shared_array.append(pi*element)
    gc.collect()
    sema.release()

# example_ar = np.arange(1, 1000) # 可正常运行
example_ar = np.arange(1, 10000) # 触发错误

# 并行逻辑
start = time.time()
# 初始化Manager与共享数据结构
manager = Manager()
shared_ar = manager.list()
# 创建信号量，数量为物理核心数（即cpu_count的1/2）
sema = Semaphore(cpu_count()//2)
job = []
# 遍历所有元素启动进程
for e in example_ar:
    sema.acquire()
    p = Process(target=do_work, args=(e, shared_ar, sema))
    job.append(p)
    p.start()
_ = [p.join() for p in job]
end_par = time.time()

# 等效串行逻辑
single_ar = []
for e in example_ar:
    single_ar.append(pi*e)
end_single = time.time()

print(f'并行耗时 {end_par-start} 秒，结果={sum(list(shared_ar))}')
print(f'串行耗时 {end_single-end_par} 秒，结果={sum(single_ar)}')

1. 进程资源未及时回收：代码将所有启动的Process对象存入job列表，直到10000个任务全部启动后才统一调用.join()。哪怕子进程已执行完毕，父进程中对应Process对象持有的文件描述符（用于进程间通信、传递退出状态）不会被释放，累计数量很快超过1024的系统限制。信号量仅限制同时运行的进程数，无法回收已退出进程的残留资源。
2. 进程创建粒度不合理：单个任务仅执行pi*element的极简运算，创建进程的开销远大于任务本身，大量进程与共享Manager的连接会额外占用文件描述符。
3. 跨库混用组件：multiprocess是第三方兼容库，和原生multiprocessing的信号量、进程对象混合使用时，可能出现资源释放逻辑不兼容的问题。

修复后代码 #

import time
import numpy as np
from math import pi
# 统一使用multiprocessing组件，如需Windows兼容可替换为multiprocess
from multiprocessing import Manager, Pool, cpu_count

def do_work(element, shared_array):
    shared_array.append(pi * element)

if __name__ == '__main__':
    example_ar = np.arange(1, 10000) # 大任务量下可正常运行

    start = time.time()
    # 初始化共享数据结构
    with Manager() as manager:
        shared_ar = manager.list()
        # 自适应进程数：取cpu_count的1/2，与原逻辑保持一致
        process_num = cpu_count() // 2
        # 进程池自动复用进程、回收资源
        with Pool(processes=process_num) as pool:
            # starmap支持向目标函数传递多参数
            pool.starmap(do_work, [(e, shared_ar) for e in example_ar])
        end_par = time.time()

        # 等效串行逻辑
        single_ar = [pi * e for e in example_ar]
        end_single = time.time()

        print(f'并行耗时 {end_par-start} 秒，结果={sum(list(shared_ar))}')
        print(f'串行耗时 {end_single-end_par} 秒，结果={sum(single_ar)}')

修复点说明 #

• 替换手动创建进程为Pool进程池：自动复用固定数量的进程，无需创建上万个进程对象，同时自动回收所有进程资源，从根源避免文件描述符泄漏
• 移除手动信号量逻辑：进程池本身已限制最大并发数，无需额外信号量控制
• 完全满足需求：
  1. 仍使用Manager().list()实现共享数据结构
  2. 通过starmap传递多参数
  3. 直接通过cpu_count()获取系统核心数，自适应调整进程池大小
• 增加if __name__ == '__main__'入口判断：避免多进程启动时的递归调用问题，适配所有系统
• 移除无效的gc.collect()调用：无资源泄漏场景下无需主动触发垃圾回收

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