Python多进程嵌套未调用join时子进程提前终止问题

1. 进程被强制回收：Web应用的请求处理进程（比如WSGI worker）处理完用户请求后，会释放关联资源。func2是这个请求进程的子进程，一旦请求结束，请求进程可能会回收所有下属子进程，导致func2和它启动的子进程还没跑完就被终止。哪怕func1没调用join，只要请求进程被回收或进入空闲清理，后台进程都会遭殃。
2. 队列逻辑乱了：原代码里queueList反复添加同一个Queue对象，循环q.get()时会重复从同一个队列取数据，不仅逻辑错误，子进程数量大时还会出现阻塞或拿到重复数据的问题。

1. 让func2彻底脱离请求进程控制 #

在Unix/Linux系统下，通过两次fork把func2变成孤儿进程，让系统的init进程接管它，这样请求进程结束也不会影响它：

import os
# ... 其他导入不变
class Test:
    def func1(self):
        # 第一次fork出子进程
        pid = os.fork()
        if pid == 0:
            # 子进程里再fork一次
            pid2 = os.fork()
            if pid2 == 0:
                # 孙子进程执行func2，之后会被init接管
                self.func2()
            # 子进程直接退出，切断和请求进程的关联
            os._exit(0)
        # 请求进程等待子进程退出，避免产生僵尸进程
        os.waitpid(pid, 0)

如果是Windows系统（不支持fork），就用默认的Process，确保daemon=False（默认就是False），同时要保证Web服务器的worker进程不会频繁重启：

def func1(self):
    p = Process(target=self.func2, daemon=False)
    p.start()
    # 不用join，让进程后台运行

2. 把队列逻辑改对 #

所有子进程共用一个Queue就行，不用创建多个队列。修改func2，确保收齐所有子进程的输出：

def func2(self):
    queue = Queue()
    jobList = []
    responseList = []
    total_jobs = 1000
    
    # 启动所有子进程
    for i in range(total_jobs):
        p = Process(target=self.work, args=(i, queue))
        p.start()
        jobList.append(p)
    
    # 从队列取total_jobs次数据，确保收齐所有子进程的输出
    for _ in range(total_jobs):
        responseList.append(queue.get())
    
    # 等待所有子进程执行完成
    for p in jobList:
        p.join()
    
    # 保存数据
    signalPath = 'path_to_somewhere/testProcess.json'
    with open(signalPath, 'w') as fileHandle:
        json.dump(responseList, fileHandle)

3. Web场景的更优方案 #

手动管理进程在Web应用里容易出问题，比如进程泄漏、资源占用失控，建议用任务队列框架（比如Celery、RQ）来管理异步任务，这是行业通用的稳定方案，比自己写进程逻辑靠谱得多。

import os
from multiprocessing import Process, Queue
import json

class Test:
    def func1(self):
        # Unix/Linux环境下的孤儿进程方案
        pid = os.fork()
        if pid == 0:
            pid2 = os.fork()
            if pid2 == 0:
                self.func2()
            os._exit(0)
        os.waitpid(pid, 0)

    def func2(self):
        queue = Queue()
        jobList = []
        responseList = []
        total_jobs = 1000
        
        for i in range(total_jobs):
            p = Process(target=self.work, args=(i, queue))
            p.start()
            jobList.append(p)
        
        for _ in range(total_jobs):
            responseList.append(queue.get())
        
        for p in jobList:
            p.join()
        
        signalPath = 'path_to_somewhere/testProcess.json'
        with open(signalPath, 'w') as fileHandle:
            json.dump(responseList, fileHandle)
    
    def work(self, i, queue):
        print(i)
        queue.put(i)
        
if __name__ == '__main__':
    classObj = Test()
    classObj.func1()

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