Python多进程Pool Queue函数式正常但OOP无输出问题排查

问题背景 #

学习Python多进程时，尝试用worker pool管理下载任务，发现函数式实现的进程池队列能正常输出，但转为OOP实现后无预期输出，确认问题与OOP实现相关，但不清楚具体原因。

可正常运行的函数式代码 #

import multiprocessing

def worker(name, que):
    que.put("%d is done" % name)
    
if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=3)
    m = multiprocessing.Manager()
    q = m.Queue(maxsize=10)
    
    pool.apply_async(worker, (33, q))
    pool.apply_async(worker, (40, q))
    pool.apply_async(worker, (27, q))
    
    while True:
        try:
            print(q.get(False))
        except:
            pass

函数式代码输出

27 is done
33 is done
40 is done

无输出的OOP实现代码 #

import multiprocessing

class Workers:
    def __init__(self):
        self.pool = multiprocessing.Pool(processes=3)
        self.m = multiprocessing.Manager()
        self.q = self.m.Queue()
        
        self.pool.apply_async(self.worker, (33, self.q))
        self.pool.apply_async(self.worker, (40, self.q))
        self.pool.apply_async(self.worker, (27, self.q))

        while True:
            try:
                print(self.q.get(False))
            except:
                pass

    def worker(self, name, que):
        que.put("%d is done" % name)


if __name__ == "__main__":
    w = Workers()

OOP代码输出

（无任何输出）

原因分析 #

问题核心在于进程池调用类方法的序列化机制：

1. 用pool.apply_async调用类实例方法self.worker时，Python需要序列化整个Workers实例（实例方法依赖实例本身）。
2. 在Windows或使用spawn启动方式的系统中，子进程会重新导入主模块，此时__init__方法会被再次执行，创建新的进程池、Manager和队列，子进程实际操作的是这个新队列，而非主进程的队列。
3. 主进程的while True循环写在__init__里，会阻塞实例初始化；子进程因重新执行__init__也陷入无限循环，无法真正执行worker方法。

解决方案 #

方案1：将worker改为静态方法 #

静态方法无需依赖实例，避免序列化整个对象的问题：

import multiprocessing

class Workers:
    def __init__(self):
        self.pool = multiprocessing.Pool(processes=3)
        self.m = multiprocessing.Manager()
        self.q = self.m.Queue()
        
        self.pool.apply_async(Workers.worker, (33, self.q))
        self.pool.apply_async(Workers.worker, (40, self.q))
        self.pool.apply_async(Workers.worker, (27, self.q))

        # 等待所有子进程完成任务
        self.pool.close()
        self.pool.join()
        
        while not self.q.empty():
            try:
                print(self.q.get(False))
            except:
                pass

    @staticmethod
    def worker(name, que):
        que.put("%d is done" % name)


if __name__ == "__main__":
    w = Workers()

方案2：将worker方法移到类外 #

保持函数式实现的特性，避免实例序列化问题：

import multiprocessing

def worker(name, que):
    que.put("%d is done" % name)

class Workers:
    def __init__(self):
        self.pool = multiprocessing.Pool(processes=3)
        self.m = multiprocessing.Manager()
        self.q = self.m.Queue()
        
        self.pool.apply_async(worker, (33, self.q))
        self.pool.apply_async(worker, (40, self.q))
        self.pool.apply_async(worker, (27, self.q))

        self.pool.close()
        self.pool.join()
        
        while not self.q.empty():
            try:
                print(self.q.get(False))
            except:
                pass


if __name__ == "__main__":
    w = Workers()

方案3：使用Process而非Pool（适合简单场景） #

直接创建进程，避免池化带来的序列化问题：

import multiprocessing

class Workers:
    def __init__(self):
        self.m = multiprocessing.Manager()
        self.q = self.m.Queue()
        processes = []
        
        for name in [33,40,27]:
            p = multiprocessing.Process(target=self.worker, args=(name, self.q))
            processes.append(p)
            p.start()
        
        for p in processes:
            p.join()
            
        while not self.q.empty():
            print(self.q.get(False))

    def worker(self, name, que):
        que.put("%d is done" % name)


if __name__ == "__main__":
    w = Workers()

额外优化点 #

1. 原代码的while True无限循环会导致CPU空转，建议改为检查队列是否为空或设置超时等待。
2. 必须调用pool.close()和pool.join()等待所有子进程完成，确保任务全部执行后再读取队列。

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