使用多进程模块

Question

我试图在Python 2.6中使用多进程模块，但显然有些地方我没有理解。我希望下面的类通过add()方法接收的数字相加，并在get_result()方法中返回和。下面的代码打印出"0"，我希望它打印出"2"。我错过了什么？

import multiprocessing

class AdderProcess(multiprocessing.Process):

    def __init__(self):
        multiprocessing.Process.__init__(self)
        self.sum = 0
        self.queue = multiprocessing.JoinableQueue(5)
        self.daemon = True
        self.start()

    def run(self):
        while True:
            number = self.queue.get()
            self.sum += number
            self.queue.task_done()

    def add(self, number):
        self.queue.put(number)

    def get_result(self):
        self.queue.join()
        return self.sum


p = AdderProcess()
p.add(1)
p.add(1)
print p.get_result()

PS. 这个问题已经解决了。 感谢各位的回答！为了让读者更容易理解，这里提供完整的可用版本：

import multiprocessing

class AdderProcess(multiprocessing.Process):

    def __init__(self):
        multiprocessing.Process.__init__(self)
        self.sum = multiprocessing.Value('d', 0.0)
        self.queue = multiprocessing.JoinableQueue(5)
        self.daemon = True
        self.start()

    def run(self):
        while True:
            number = self.queue.get()
            self.sum.value += number
            self.queue.task_done()

    def add(self, number):
        self.queue.put(number)

    def get_result(self):
        self.queue.join()
        return self.sum.value

p = AdderProcess()
p.add(1)
p.add(1)
print p.get_result()

- Mats Ekberg

Answer 1

将 self.sum = 0 改为 self.sum = multiprocessing.Value('d', 0.0)，并使用 self.sum.value 来访问或更改值。

class AdderProcess(multiprocessing.Process):    
    def __init__(self):
        ...
        self.sum = multiprocessing.Value('d', 0.0) 
        ...
    def run(self):
        while True:
            number = self.queue.get()
            self.sum.value += number    # <-- use self.sum.value
            self.queue.task_done()
    def get_result(self):
        self.queue.join()
        return self.sum.value           # <-- use self.sum.value

问题是这样的：一旦在__init__中调用self.start()，主进程就会分叉出一个子进程。所有值都被复制。现在有两个版本的p。在主进程中，p.sum为0。在子进程中，调用run方法并将p.sum增加到2。但是当主进程调用p.get_result()时，它的p版本仍然具有p.sum等于0的值。因此会打印出0。

当您想要在进程之间共享浮点值时，需要使用共享机制，例如mp.Value。

有关如何共享值的更多选项，请参见“在进程之间共享状态”。

- unutbu

---

假设你的意思是让我用你的代码替换我的初始化程序，但是当执行该代码时会导致TypeError异常。它对你有效吗？你能详细解释一下你在这里做什么吗？ - Mats Ekberg

抱歉，我确实运行了代码，只是忘记包含所有更改。您还需要将 self.sum 更改为 self.sum.value，以访问或更改其值。 - unutbu

好的，我现在明白了。没有RPC魔法可以使方法调用到达另一个进程，唯一共享的信息是显式共享的，用于此目的提供的数据类型。我想这很有道理。谢谢！ - Mats Ekberg

Answer 2

self.sum 是 2… 在这个过程中:

def run(self):
    while True:
        number = self.queue.get()
        print "got %s from queue" % number
        print "Before adding - self.sum = %d" % self.sum
        self.sum += number
        print "After adding - self.sum = %d" % self.sum
        self.queue.task_done()

[ 13:56 jon@host ~ ]$ ./mp.py
got 1 from queue
Before adding - self.sum = 0
After adding - self.sum = 1
got 1 from queue
Before adding - self.sum = 1
After adding - self.sum = 2

请参考 多进程16.3.1.4 - 在进程之间共享状态 来了解如何使不同的进程中的self.sum保持一致。

- chown