python子类化多处理.Process

Question

我是python面向对象的新手，我正在将现有的应用程序重写为面向对象的版本，因为现在开发人员越来越多，我的代码也变得不可维护。

通常我使用多处理队列，但我从这个示例中发现可以将http://www.doughellmann.com/PyMOTW/multiprocessing/basics.html划分为multiprocessing.Process子类，所以我认为这是一个好主意，我编写了一个类来进行如下测试：

代码：

from multiprocessing import Process
class Processor(Process):
    def return_name(self):
        return "Process %s" % self.name
    def run(self):
        return self.return_name()

processes = []


if __name__ == "__main__":

        for i in range(0,5):
                p=Processor()
                processes.append(p)
                p.start()
        for p in processes:
                p.join()

但是，我无法取回值，如何使用队列？

编辑：我想得到返回值，并想把Queues()放在哪里。

Answer 1

Process.run的返回值不在任何地方。您需要将它们发送回父进程，例如使用multiprocessing.Queue（docs here）。

Answer 2

谢谢大家。

现在我是这样做的：）

在这个例子中，我使用了多个queus，因为我不想在每个ohter之间通信，而只想与父进程通信。

from multiprocessing import Process,Queue
class Processor(Process):
    def __init__(self,queue):
        Process.__init__(self)
        self.que=queue
    def get_name(self):
        return "Process %s" % self.name
    def run(self):
        self.que.put(self.get_name())



if __name__ == "__main__":

        processes = []
        for i in range(0,5):
                p=Processor(Queue())
                processes.append(p)
                p.start()
        for p in processes:
                p.join()
                print p.que.get()

Answer 3

子类化multiprocessing.Process：

However I cannot get back the values, how can I use queues in this way?

进程需要Queue()来接收结果。。。下面是一个如何对multiprocessing.Process进行子类划分的示例。。。

from multiprocessing import Process, Queue
class Processor(Process):

    def __init__(self, queue, idx, **kwargs):
        super(Processor, self).__init__()
        self.queue = queue
        self.idx = idx
        self.kwargs = kwargs

    def run(self):
        """Build some CPU-intensive tasks to run via multiprocessing here."""
        hash(self.kwargs) # Shameless usage of CPU for no gain...

        ## Return some information back through multiprocessing.Queue
        ## NOTE: self.name is an attribute of multiprocessing.Process
        self.queue.put("Process idx={0} is called '{1}'".format(self.idx, self.name))

if __name__ == "__main__":
    NUMBER_OF_PROCESSES = 5

    ## Create a list to hold running Processor object instances...
    processes = list()

    q = Queue()  # Build a single queue to send to all process objects...
    for i in range(0, NUMBER_OF_PROCESSES):
        p=Processor(queue=q, idx=i)
        p.start()
        processes.append(p)

    # Incorporating ideas from this answer, below...
    #    https://stackoverflow.com/a/42137966/667301
    [proc.join() for proc in processes]
    while not q.empty():
        print "RESULT: {0}".format(q.get())   # get results from the queue...

在我的机器上，这会导致。。。

$ python test.py
RESULT: Process idx=0 is called 'Processor-1'
RESULT: Process idx=4 is called 'Processor-5'
RESULT: Process idx=3 is called 'Processor-4'
RESULT: Process idx=1 is called 'Processor-2'
RESULT: Process idx=2 is called 'Processor-3'
$

使用multiprocessing.Pool：

FWIW，我发现对multiprocessing.Process进行子类化的一个缺点是，如果您不需要生产者和消费者代码通过队列相互通信，就不能充分利用multiprocessing.Pool的所有内置优点，Pool为您提供了一个非常好的API。

你可以做很多只是一些创造性的回报价值观。。。在下面的示例中，我使用dict()封装来自pool_job()的输入和输出值。。。

from multiprocessing import Pool

def pool_job(input_val=0):
    # FYI, multiprocessing.Pool can't guarantee that it keeps inputs ordered correctly
    # dict format is {input: output}...
    return {'pool_job(input_val={0})'.format(input_val): int(input_val)*12}

pool = Pool(5)  # Use 5 multiprocessing processes to handle jobs...
results = pool.map(pool_job, xrange(0, 12)) # map xrange(0, 12) into pool_job()
print results

这将导致：

[
    {'pool_job(input_val=0)': 0}, 
    {'pool_job(input_val=1)': 12}, 
    {'pool_job(input_val=2)': 24}, 
    {'pool_job(input_val=3)': 36}, 
    {'pool_job(input_val=4)': 48}, 
    {'pool_job(input_val=5)': 60}, 
    {'pool_job(input_val=6)': 72}, 
    {'pool_job(input_val=7)': 84}, 
    {'pool_job(input_val=8)': 96}, 
    {'pool_job(input_val=9)': 108}, 
    {'pool_job(input_val=10)': 120}, 
    {'pool_job(input_val=11)': 132}
]

显然，在pool_job()中还有很多其他的改进，比如错误处理，但是这说明了一些要点。仅供参考，this answer提供了另一个如何使用multiprocessing.Pool的示例。