使用 Python 的多处理模块生成类属性答案

【问题标题】：Using Python's Multiprocessing Module To Generate Class Attributes使用 Python 的多处理模块生成类属性
【发布时间】：2014-10-21 19:05:21
【问题描述】：

我是多处理模块的新手，想知道一个类是否有可能产生可以与之通信的工作类。我有一个对象接收数据的实例，并且需要根据该数据快速构造属性，这些属性本身就是对象。下面的代码是一个简单的例子：

class Worker(multiprocessing.Process):

    def __init__(self, in_queue, out_queue):
        multiprocessing.Process.__init__(self)
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):

        #Loop through the in_queue until a None is found
        for tup in iter(self.in_queue.get,None):

            #Try to manipulate the data
            try:
                self.out_queue.put(tup[0]*tup[1])
            except:
                self.out_queue.put('Error')
            self.in_queue.task_done()

        #Remove the None from the in_queue
        self.in_queue.task_done()

class Master():

    def __init__(self,data):

        #Initialize some data to be operated on
        self.data=data

    def gen_attributes(self):

        #Initialize Queues to interact with the workers
        in_queue=multiprocessing.JoinableQueue()
        out_queue=multiprocessing.Queue()

        #Create workers to operate on the data
        for i in range(4):
            Worker(in_queue,out_queue).start()

        #Add data to the input queue
        for tup in self.data:
            in_queue.put(tup)        

        #Stop Condition for each worker
        for _ in range(4):
            in_queue.put(None)
        in_queue.close()

        #Wait for processes to finish
        in_queue.join()

        #Store the output as new attributes of the class
        self.attributes=[]
        for _ in range(0,out_queue.qsize()):
            self.attributes.append(out_queue.get()) 

#Create and instance of the Master class, and have it generate it's own attributes using the multiprocessing module
data=[(1,2),(2,2),(3,4)]
M=Master(data)
M.gen_attributes()
print M.attributes

本质上，Master 类的实例是使用给定数据生成的。然后，主类将该数据传递给多个工作人员进行操作并放入输出队列中。 Master 类然后使用该输出来分配自己的属性。

【问题讨论】：

标签： python multiprocessing

【解决方案1】：

这看起来是multiprocessing.Pool 的完美用例。您可能会注意到您的程序挂起并且Master 没有收到您期望的任何属性。那是因为它当时还没有收到任何信息，因为它需要阻止out_queue 才能从队列中读取信息，因为在您的程序从out_queue 读取信息时它是空的。

一个简单的解决方法是在队列的get 方法上阻塞并等待超时，如下所示：

 while True:
  attr = out_queue.get(True, 0.1)
  if not attr:
    break
  self.attributes.append(attr)

这可行，但它不是一个干净的解决方案。您可以对其进行修改，并对其进行试验以获得您想要的结果，我建议使用哨兵值。

【讨论】：

肖恩，谢谢您的回复。这很有趣。我以为它挂在 in_queue.join()... 即工人没有做他们的工作。但是如果是传入in_queue.join()，那是不是说明所有的job都完成了，out_queue也应该满是结果了？我还尝试检查所有工人也加入了，并得到与上述相同的问题。
肖恩，我也实施了您的解决方案，但我似乎无法让它发挥作用。你有实现吗？
我有，但它在我的个人笔记本电脑上。我下班后会尝试发布。

【解决方案2】：

想通了。在 if __name__ == '__main__' 下封装 Master 类的实例化解决了这个问题。显然，这是一个专门的 Windows 问题。更多信息在这里：

https://docs.python.org/2/library/multiprocessing.html#multiprocessing-programming

【讨论】：