适用于并行进程的Python多处理

Question

如果对某些人来说这太简单了，我感到抱歉，但是我仍然不了解python的多处理技巧。我已经阅读了
http://docs.python.org/dev/library/multiprocessing


http://pymotw.com/2/multiprocessing/basics.html
以及Google给我的许多其他教程和示例…其中许多从这里也是。

好吧，我的情况是我必须计算许多numpy矩阵，然后需要将它们存储在单个numpy矩阵中。假设我要使用20个核心（或者我可以使用20个核心），但是由于资源使进程一直存活到“死亡”，因此我没有成功使用池资源。所以我想做这样的事情：

from multiprocessing import Process, Queue  
import numpy as np

def f(q,i):  
     q.put( np.zeros( (4,4) ) )

if __name__ == '__main__':   
     q = Queue()   
     for i in range(30):   
          p = Process(target=f, args=(q,))  
          p.start()  
          p.join()  
     result = q.get()  
     while q.empty() == False:
          result += q.get()  
     print result

但是然后看来这些进程不是并行运行，而是按顺序运行（如果我错了，请纠正我），而且我不知道它们在执行计算后是否会死亡（因此对于20多个进程，尽了自己的本分，让核心可以自由地用于其他过程）。另外，对于一个非常大的数字（比如说100.000），将所有这些矩阵（可能也真的很大）存储在队列中将占用大量内存，使代码无用，因为这种想法是将每个结果都放在每次迭代中在最终结果中，就像使用锁（及其锁（及其Acquisition（）和release（）方法））一样，但是如果此代码不用于并行处理，则锁也无济于事…

我希望有人可以帮助我。

提前致谢！

Answer 1

您是正确的，它们在您的示例中按顺序执行。

p.join()导致当前线程阻塞，直到完成执行。您可能想在for循环之外单独加入您的进程（例如，通过将它们存储在列表中，然后对其进行迭代），或使用类似numpy.Pool和apply_async的回调函数。这样一来，您也可以直接将其添加到结果中，而无需保留对象。

例如：

def f(i):  
    return i*np.identity(4)

if __name__ == '__main__':
    p=Pool(5)
    result = np.zeros((4,4))
    def adder(value):
        global result
        result += value

    for i in range(30):
        p.apply_async(f, args=(i,), callback=adder)
    p.close()
    p.join()
    print result

最后关闭并再加入池可确保池的进程已完成，并且result对象的计算已完成。您也可以调查将其Pool.imap用作解决问题的方法。该特定解决方案如下所示：

if __name__ == '__main__':
    p=Pool(5)
    result = np.zeros((4,4))

    im = p.imap_unordered(f, range(30), chunksize=5)

    for x in im:
        result += x

    print result

这对于您的特定情况比较干净，但对于最终尝试执行的操作可能不是这样。

至于存储所有变化的结果，如果我理解您的问题，您可以将其添加到回调方法（如上）中或使用imap/
imap_unordered（同时存储结果，但每次一次）添加到结果中您会在构建时清除它）。这样一来，存储时间就不会超过添加到结果中所需的时间。