Python multiprocessing最佳实践：大处理量多进程内存不释放与用法疑问

现有代码的问题定位 #

当子进程执行完自身逻辑退出后，如果父进程没有通过系统调用回收子进程的退出状态，子进程会变成僵尸进程，持续占用进程表项、残留内核内存，直到父进程主动完成回收。你写的两段代码都没有形成完整的进程回收闭环，这是内存不释放的核心原因：

• 第一段带join()的代码：仅完成了僵尸进程的内核回收，但没有主动调用proc.close()释放父进程侧的资源。每个Process实例内部持有和子进程通信的管道、同步原语、缓存引用，不主动执行close的话，这部分父进程侧的内存不会被自动垃圾回收。
• 第二段修改后的代码：启动完全部进程后立刻遍历检查is_alive()，此时绝大多数进程仍在运行，不满足close()方法的调用前置条件（必须对已完全退出的进程调用，否则会直接抛出ValueError），完全达不到“进程跑完就立刻回收”的预期效果。

join()的实际工作逻辑 #

你觉得文档描述和实际效果不符，本质是对join的定位理解偏差：

• join()从来不是用来实现子进程之间互相等待的，它是父进程主导的回收动作：调用proc.join()时，父进程会阻塞当前执行流，直到目标子进程完全退出，操作系统将子进程的退出状态回传给父进程，完成僵尸进程的内核资源回收。
• 子进程之间是完全隔离的内存地址空间，默认没有任何通信和感知能力，本来就不存在“互相等待”的逻辑，所有的进程同步、资源回收动作都需要父进程主导完成。

手动管理Process与Pool进程池的适用差异 #

手动创建Process的局限性 #

你当前的写法是遍历到哪个文件分区就对应启动一个进程，一次性把所有任务的进程全部拉起——比如有100个分区就直接启动100个进程，完全不考虑机器的CPU核心数上限，这种写法在海量数据处理场景下有几个硬伤：

• 进程数远大于CPU核心数时，操作系统会频繁做上下文切换，整体性能不升反降
• 一次性创建过多进程，光进程初始化的开销就会吃掉多进程加速的收益
• 必须等所有进程启动完成后才能统一回收资源，内存释放滞后，进程运行的峰值内存占用极高

Pool进程池的核心优势 #

进程池的设计刚好匹配海量数据处理的场景需求：

• 池的大小固定（默认等于机器CPU核心数），不会因为任务数太多一下子启动成百上千个进程占满内存
• 内置自动资源回收逻辑：工作进程执行完任务后，池会自动完成wait回收、关闭对应进程对象的通信资源，不会残留僵尸进程
• 内置任务调度：你只需要把待处理的任务列表丢给池，池会自动把任务分配给空闲的工作进程执行，不需要你手动维护进程列表、手动判断进程运行状态
• 统一的资源释放入口：只要正确执行池的生命周期管理逻辑，就能一次性清掉池占用的所有管道、信号量、进程资源，内存占用比手动管理稳定得多

两类场景的正确代码示例 #

手动管理进程的正确写法 #

仅适合进程数量少、需要严格控制启动时机的场景，海量数据处理不推荐

for file_to_process in file_partition:
    if len(file_to_process) == 0:
        continue
    logger = self.__get_logger__(task.lower(), len(proc_list) + 1)
    proc = Process(
        target=self.__launch_thread__,
        args=(task, manager_class, file_to_process, list_action, str_target_path, str_target_buffer_path, logger)
    )
    proc.start()
    proc_list.append(proc)

# 严格按照回收顺序执行：先等所有进程退出，再逐个关闭
for proc in proc_list:
    # join完成子进程的内核资源回收
    proc.join()
    # 此时进程已完全退出，调用close释放父进程侧的Process对象资源
    proc.close()
# 清空进程列表，释放父进程侧的列表引用
proc_list.clear()

适配海量数据处理的Pool写法 #

import os
from multiprocessing import Pool

# 包装原来的任务函数，适配池的单入参要求
def task_wrapper(partition_info):
    task, manager_class, file_partition, list_action, target_path, buffer_path, task_id = partition_info
    logger = self.__get_logger__(task.lower(), task_id)
    return self.__launch_thread__(
        task, manager_class, file_partition, list_action,
        target_path, buffer_path, logger
    )

if __name__ == "__main__":
    task_list = []
    task_id = 1
    # 整理所有非空分区的任务参数
    for file_to_process in file_partition:
        if len(file_to_process) == 0:
            continue
        task_list.append((
            task, manager_class, file_to_process, list_action,
            str_target_path, str_target_buffer_path, task_id
        ))
        task_id += 1
    
    # 用with语法管理池的生命周期，自动完成资源回收
    with Pool(processes=os.cpu_count()) as pool:
        # 自动调度所有任务执行
        pool.map(task_wrapper, task_list)
    # 离开with代码块时，池会自动执行close() + join()，释放所有关联资源

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