我也曾经遇到过这个问题。举个简单的例子,我把函数作为类的数据成员:
from multiprocessing import Pool
import itertools
pool = Pool()
class Example(object):
def __init__(self, my_add):
self.f = my_add
def add_lists(self, list1, list2):
return pool.map(self.f,list1,list2)
我需要在Pool.map()调用中使用self.f函数,但self.f不接受元组作为参数。由于该函数嵌入在类中,所以我不清楚如何编写其他答案建议的包装器类型。
我通过使用一个不同的包装器解决了这个问题,该包装器接受元组/列表,其中第一个元素是函数,其余元素是该函数的参数,称为eval_func_tuple(f_args)。使用此方法,有问题的行可以替换为return pool.map(eval_func_tuple, itertools.izip(itertools.repeat(self.f), list1, list2))。以下是完整的代码:
文件: util.py
def add(a, b): return a+b
def eval_func_tuple(f_args):
"""Takes a tuple of a function and args, evaluates and returns result"""
return f_args[0](*f_args[1:])
文件:main.py
from multiprocessing import Pool
import itertools
import util
pool = Pool()
class Example(object):
def __init__(self, my_add):
self.f = my_add
def add_lists(self, list1, list2):
return pool.map(util.eval_func_tuple,
itertools.izip(itertools.repeat(self.f), list1, list2))
if __name__ == '__main__':
myExample = Example(util.add)
list1 = [1, 2, 3]
list2 = [10, 20, 30]
print myExample.add_lists(list1, list2)
运行main.py将得到[11, 22, 33]。可以随意改进,例如eval_func_tuple也可以修改为接受关键字参数。
另外,在其他答案中,“parmap”函数可以在进程数量超过可用CPU数量的情况下更加高效。我在下面复制了一个编辑过的版本。这是我的第一篇帖子,我不确定是否应该直接编辑原始答案。我还重命名了一些变量。
from multiprocessing import Process, Pipe
from itertools import izip
def spawn(f):
def fun(pipe,x):
pipe.send(f(x))
pipe.close()
return fun
def parmap(f,X):
pipe=[Pipe() for x in X]
processes=[Process(target=spawn(f),args=(c,x)) for x,(p,c) in izip(X,pipe)]
numProcesses = len(processes)
processNum = 0
outputList = []
while processNum < numProcesses:
endProcessNum = min(processNum+multiprocessing.cpu_count(), numProcesses)
for proc in processes[processNum:endProcessNum]:
proc.start()
for proc in processes[processNum:endProcessNum]:
proc.join()
for proc,c in pipe[processNum:endProcessNum]:
outputList.append(proc.recv())
processNum = endProcessNum
return outputList
if __name__ == '__main__':
print parmap(lambda x:x**x,range(1,5))
IPython.Parallel
中也遇到了闭包的类似问题,但那里可以通过将对象推送到节点来解决该问题。在多进程中解决这个问题似乎相当麻烦。 - Alex Scalculate
是可被 pickle 序列化的,因此似乎可以通过以下步骤解决:1)创建一个函数对象,其构造函数复制了一个calculate
实例;2)将该函数对象的实例传递给Pool
的map
方法。是这样吗? - rd11multiprocessing
模块的一些限制是由于其旨在成为跨平台实现,以及 Windows 中缺少类似fork(2)
的系统调用所致。如果您不关心 Win32 支持,则可能有一个更简单的基于进程的解决方法。或者,如果您准备使用线程而不是进程,则可以将from multiprocessing import Pool
替换为from multiprocessing.pool import ThreadPool as Pool
。 - Aya