python - multiprocessing.Queue 不能作为类中的实例变量工作？

Question

我想将多处理任务封装到一个类中。控制函数和工作函数都是该类的成员。这些工作线程使用 Pool.map_async() 运行，因此可以在其他工作线程仍在运行时处理结果。处理结果存储在multiprocessing.Queue中。当Queue是实例变量时，它不起作用，而全局变量或类变量，它起作用。

示例:

import multiprocessing 
class A():
    # Queue as instance variable
    def __init__(self):
        self.qout = multiprocessing.Queue()
    def worker(self,x):
        self.qout.put(x*x)   
    def process(self):
        values = range(10)
        with multiprocessing.Pool() as pool:
            res = pool.map_async(self.worker,values)        
            while (not self.qout.empty() or
                not res.ready()):
                val = self.qout.get()
                print(val)

qoutB = multiprocessing.Queue()
class B():
    # Queue as global variable
    def __init__(self):
        pass   
    def worker(self,x):
        qoutB.put(x*x)       
    def process(self):
        values = range(10)       
        with multiprocessing.Pool() as pool:
            res = pool.map_async(self.worker,values)           
            while (not qoutB.empty() or
                not res.ready()):
                val = qoutB.get()
                print(val)

class C():
    # Queue as Class variable
    qout = multiprocessing.Queue()
    def __init__(self):
        pass
    def worker(self,x):
        self.qout.put(x*x)   
    def process(self):
        values = range(10)
        with multiprocessing.Pool() as pool:
            res = pool.map_async(self.worker,values)        
            while (not self.qout.empty() or
                not res.ready()):
                val = self.qout.get()
                print(val)  

现在，当您按如下方式调用类时(将其放在类定义下方)

a=A()
a.process()

不起作用(可能停止等待self.qout.get()，但是

a=B()
a.process()

和

a=C()
a.process()

有效(打印结果)。为什么？

我在Python documentation中没有找到任何相关信息。我没有尝试将队列作为参数传递，但它是一个应该对用户隐藏的功能。

B 选项应该是没有问题的，C 并不理想，因为队列将在类的所有实例之间共享。

注意:这是在 Linux(Debian、来自存储库的 Python 3.5)上测试的。

Answer 1

再说一遍，这不是您问题的答案。不过，我发布它是因为它使整个问题变得毫无意义 - 因为您实际上不需要显式创建和使用 multiprocessing.Queue 来执行类似的操作。

请考虑使用 concurrent.futures.ProcessPoolExecutor完成任务。

例如:

import concurrent.futures

class A_Prime():
    def __init__(self):
        pass

    def worker(self, x):
        return x*x

    def process(self):
        with concurrent.futures.ProcessPoolExecutor() as executor:
            classname = type(self).__name__
            print(classname, '- calling executor.map')
            res = [value for value in executor.map(self.worker, range(10))]
            print(classname, '- executor.map finished')
            print('  result:', res)


if __name__ == '__main__':
    test = A_Prime()
    test.process()
    print('done')

输出:

A_Prime - calling executor.map
A_Prime - executor.map finished
  result: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
done