如果对某些人来说这太简单了,我感到抱歉,但是我仍然不了解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()方法))一样,但是如果此代码不用于并行处理,则锁也无济于事…
我希望有人可以帮助我。
提前致谢!
您是正确的,它们在您的示例中按顺序执行。
p.join()导致当前线程阻塞,直到完成执行。您可能想在for循环之外单独加入您的进程(例如,通过将它们存储在列表中,然后对其进行迭代),或使用类似numpy.Pool和apply_async的回调函数。这样一来,您也可以直接将其添加到结果中,而无需保留对象。
p.join()
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用作解决问题的方法。该特定解决方案如下所示:
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(同时存储结果,但每次一次)添加到结果中您会在构建时清除它)。这样一来,存储时间就不会超过添加到结果中所需的时间。
imap
imap_unordered