Python多进程处理大量任务中途停止问题求助

问题核心 #

你采用手动分批启动Process的方式处理10万级文件时，程序会在join步骤中途卡住停止，核心问题出在进程管理逻辑错误和手动维护进程/队列的复杂度上。

错误分析 #

1. __name__ == "__main__"位置错误：你把这个判断写在了main函数内部，Windows系统下子进程启动时会重新导入脚本，导致子进程也会执行main函数内的逻辑，引发进程创建混乱，这是多进程代码的常见陷阱。
2. Queue阻塞风险：默认的multiprocessing.Queue有固定缓冲区大小，若子进程持续往Queue中写入数据，而主进程未及时读取，缓冲区满后子进程会阻塞在put操作上，导致进程无法结束，join步骤永久卡住。
3. 手动进程管理的资源泄漏：即使分批启动，手动创建、启动、join、close的流程容易遗漏异常处理，比如某个进程崩溃后未被正确回收，累积后会导致系统资源耗尽，程序停止。

修正方案：用Pool简化进程管理 #

推荐直接使用multiprocessing.Pool，它会自动维护固定数量的工作进程，复用进程资源，还内置了可靠的结果收集机制，比手动管理Process+Queue高效且不易出错。

修正后的代码

from multiprocessing import Pool
import time

# 模拟文件处理函数，实际替换为你的读文件+解析逻辑
def process_file(file_id):
    # 此处替换为真实的文件读取与计算逻辑
    time.sleep(0)  # 模拟处理耗时
    return file_id  # 返回处理结果

def main(NProcs):
    FILES_TO_READ = [i for i in range(100000)]
    start = time.time()

    with Pool(NProcs) as pool:
        # imap_unordered按结果完成顺序返回，比map更快（无需保持顺序时使用）
        # 需要保持原文件顺序则改用map
        results = pool.imap_unordered(process_file, FILES_TO_READ)
        
        # 实时收集结果，避免一次性占用大量内存
        Q_LISTED = []
        for res in results:
            Q_LISTED.append(res)

    end = time.time()
    delta_time = round(end - start, 2)

    print('____________')
    print('Results')
    # 仅打印前10条示例，避免输出10万条数据
    for element in Q_LISTED[:10]:
        print('\t', element)
    print(f'... 共{len(Q_LISTED)}条结果')
    
    print('\ntook', delta_time, 's')
    print('\n___________________________')
    print('  ...\n      THE END\n              ...')
    return 1

if __name__ == "__main__":
    main(10)

关键改进点 #

• Pool自动管理进程：with Pool(NProcs)会在代码块结束后自动关闭并join所有进程，无需手动处理，彻底避免资源泄漏。
• 规避Queue阻塞：Pool的imap_unordered/map自动处理结果传递，底层采用更可靠的管道机制，不会出现缓冲区满导致的阻塞问题。
• 正确的__name__判断：将if __name__ == "__main__"放在最外层，确保子进程启动时不会重复执行主逻辑，这是Windows下多进程代码的必要条件。
• 进程资源复用：Pool会复用工作进程，避免频繁创建销毁进程的开销，处理10万级任务时效率更高。

额外建议 #

1. 若文件处理逻辑占用大量内存，建议将NProcs设置为CPU核心数的1-2倍（如8核CPU设为8或16），避免内存耗尽。
2. 实际处理文件时，可直接给process_file传入文件路径而非ID，让子进程完成文件读取和解析，减少主进程与子进程的数据传递开销。
3. 需要实时监控进度时，可在for res in results循环中加入进度打印，比如每处理1000个文件打印一次进度。

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