我试图使用multiprocessing的Pool.map()函数来同时划分工作。当我使用以下代码时,它工作得很好:

import multiprocessing

def f(x):
    return x*x

def go():
    pool = multiprocessing.Pool(processes=4)        
    print pool.map(f, range(10))


if __name__== '__main__' :
    go()

然而,当我在更面向对象的方法中使用它时,它就不起作用了。它给出的错误信息是:

PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup
__builtin__.instancemethod failed

这发生时,以下是我的主程序:

import someClass

if __name__== '__main__' :
    sc = someClass.someClass()
    sc.go()

下面是我的someClass类:

import multiprocessing

class someClass(object):
    def __init__(self):
        pass

    def f(self, x):
        return x*x

    def go(self):
        pool = multiprocessing.Pool(processes=4)       
        print pool.map(self.f, range(10))

有人知道问题是什么吗,或者有什么简单的解决方法吗?


当前回答

上面parisjohn的解决方案对我很有效。此外,代码看起来很干净,易于理解。在我的例子中,有几个函数要使用Pool调用,所以我在下面修改了parisjohn的代码。我使__call__能够调用几个函数,函数名在go()的参数dict中传递:

from multiprocessing import Pool
class someClass(object):
    def __init__(self):
        pass
    
    def f(self, x):
        return x*x
    
    def g(self, x):
        return x*x+1    

    def go(self):
        p = Pool(4)
        sc = p.map(self, [{"func": "f", "v": 1}, {"func": "g", "v": 2}])
        print sc

    def __call__(self, x):
        if x["func"]=="f":
            return self.f(x["v"])
        if x["func"]=="g":
            return self.g(x["v"])        

sc = someClass()
sc.go()

其他回答

你也可以在someClass()中定义__call__()方法,该方法调用someClass.go(),然后将someClass()的一个实例传递给池。这个对象是可pickle的,它工作得很好(为我)…

上面parisjohn的解决方案对我很有效。此外,代码看起来很干净,易于理解。在我的例子中,有几个函数要使用Pool调用,所以我在下面修改了parisjohn的代码。我使__call__能够调用几个函数,函数名在go()的参数dict中传递:

from multiprocessing import Pool
class someClass(object):
    def __init__(self):
        pass
    
    def f(self, x):
        return x*x
    
    def g(self, x):
        return x*x+1    

    def go(self):
        p = Pool(4)
        sc = p.map(self, [{"func": "f", "v": 1}, {"func": "g", "v": 2}])
        print sc

    def __call__(self, x):
        if x["func"]=="f":
            return self.f(x["v"])
        if x["func"]=="g":
            return self.g(x["v"])        

sc = someClass()
sc.go()

甚至不需要安装完整的悲情包。

实际上,唯一需要的包是dill (pip install dill),然后用dill覆盖多处理Pickler:

dill.Pickler.dumps, dill.Pickler.loads = dill.dumps, dill.loads
multiprocessing.reduction.ForkingPickler = dill.Pickler
multiprocessing.reduction.dump = dill.dump

这个答案来自https://stackoverflow.com/a/69253561/10686785

问题是,多进程必须pickle对象,以便在进程之间悬挂它们,而绑定的方法是不可pickle的。解决方法(不管您是否认为它“简单”;-)是将基础设施添加到程序中,以允许对此类方法进行pickle,将其注册到copy_reg标准库方法中。

例如,Steven Bethard对这个线程的贡献(接近线程的末尾)展示了一个通过copy_reg允许方法pickle / unpickle的完美可行的方法。

所有这些解决方案都是丑陋的,因为多处理和酸洗是坏的和有限的,除非跳出标准库。

如果你使用一个叫做pathos的多处理分支。在multiprocessing的map函数中,可以直接使用类和类方法。这是因为使用了dill而不是pickle或cPickle,并且dill可以序列化python中的几乎所有内容。

感伤。Multiprocessing还提供了一个异步映射函数…它可以映射具有多个参数的函数(例如map(math. map))。Pow, [1,2,3], [4,5,6])))

看到的: 多处理和莳萝一起可以做什么?

和: http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization/

>>> import pathos.pools as pp
>>> p = pp.ProcessPool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> p.map(t.plus, x, y)
[4, 6, 8, 10]

明确地说,你可以做你一开始想做的,如果你想的话,你可以通过解释器来做。

>>> import pathos.pools as pp
>>> class someClass(object):
...   def __init__(self):
...     pass
...   def f(self, x):
...     return x*x
...   def go(self):
...     pool = pp.ProcessPool(4)
...     print pool.map(self.f, range(10))
... 
>>> sc = someClass()
>>> sc.go()
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
>>> 

在这里获取代码: https://github.com/uqfoundation/pathos