如何使用Python joblib并行调用类方法？报错问题求解

你遇到的问题其实有两个核心点：一是语法错误导致无法序列化生成器，二是多进程的内存隔离特性让子进程修改的实例属性没法同步回主进程。咱们一步步拆解解决：

先看错误根源 #

你运行代码后触发的错误：

Traceback (most recent call last):
File "", line 1, in runfile('G:/My Drive/CODE/stackoverflow/parallel_classfunc/parallel_classfunc.py', wdir='G:/My Drive/CODE/stackoverflow/parallel_classfunc')
File "C:\ProgramData\Anaconda2\lib\site-packages\spyder\utils\site\sitecustomize.py", line 710, in runfile execfile(filename, namespace)
File "C:\ProgramData\Anaconda2\lib\site-packages\spyder\utils\site\sitecustomize.py", line 86, in execfile exec(compile(scripttext, filename, 'exec'), glob, loc)
File "G:/My Drive/CODE/stackoverflow/parallel_classfunc/parallel_classfunc.py", line 12, in Parallel(n_jobs=4)(delayed(run.p() for run in runs))
File "C:\ProgramData\Anaconda2\lib\site-packages\joblib\parallel.py", line 183, in delayed pickle.dumps(function)
File "C:\ProgramData\Anaconda2\lib\copy_reg.py", line 70, in _reduce_ex raise TypeError, "can't pickle %s objects" % base.name
TypeError: can't pickle generator objects

这个错误来自两处写法问题：

1. delayed()的参数必须是可调用对象（比如函数、类方法），但你直接写了run.p()——这会立刻执行方法，还把整个生成器对象丢给了delayed，而joblib无法序列化生成器，所以抛出了异常。
2. 就算修正了语法，joblib默认用多进程模式（Windows下强制多进程），子进程会复制一份实例，修改的只是子进程里的副本，主进程的原实例完全不会被更新，最后打印run.y还是会报错。

解决方案1：多进程模式下返回实例并重新赋值 #

既然子进程的修改没法同步回主进程，我们可以让类方法执行完后返回整个实例，再用Parallel的执行结果替换原来的实例列表。修改后的代码如下：

from joblib import Parallel, delayed

class A(object):
    def __init__(self, x):
        self.x = x
    def p(self):
        self.y = self.x**2
        return self  # 执行完成后返回自身实例

if __name__ == '__main__':
    runs = [A(x) for x in range(20)]
    # 正确的delayed用法：传入方法和实例，或者直接传入绑定了实例的方法
    runs = Parallel(n_jobs=4)(delayed(A.p)(run) for run in runs)
    # 也可以写成：delayed(run.p)() for run in runs
    for run in runs:
        print(run.y)

执行后，runs里的每个实例都是子进程中修改过的副本，就能正常访问y属性了。

解决方案2：改用多线程模式（仅适用于IO密集型任务） #

如果你的类方法是IO密集型（比如读写文件、网络请求），可以让joblib使用多线程而非多进程——线程共享同一内存空间，修改的就是原实例的属性。只需要给Parallel添加backend='threading'参数：

from joblib import Parallel, delayed

class A(object):
    def __init__(self, x):
        self.x = x
    def p(self):
        self.y = self.x**2

if __name__ == '__main__':
    runs = [A(x) for x in range(20)]
    # 切换为线程池，同时修正delayed的语法
    Parallel(n_jobs=4, backend='threading')(delayed(run.p)() for run in runs)
    for run in runs:
        print(run.y)

⚠️ 注意：如果是CPU密集型任务，多线程因为Python GIL的存在，不会提升运行速度，甚至可能更慢，这种情况还是用方案1的多进程+返回实例的方式更合适。

更优替代方案：使用标准库concurrent.futures #

如果你觉得joblib的写法有点繁琐，Python标准库的concurrent.futures也能实现同样的功能，写法更直观：

多进程版（CPU密集型推荐） #

from concurrent.futures import ProcessPoolExecutor

class A(object):
    def __init__(self, x):
        self.x = x
    def p(self):
        self.y = self.x**2
        return self

if __name__ == '__main__':
    runs = [A(x) for x in range(20)]
    with ProcessPoolExecutor(max_workers=4) as executor:
        runs = list(executor.map(lambda obj: obj.p(), runs))
    for run in runs:
        print(run.y)

多线程版（IO密集型推荐） #

from concurrent.futures import ThreadPoolExecutor

class A(object):
    def __init__(self, x):
        self.x = x
    def p(self):
        self.y = self.x**2

if __name__ == '__main__':
    runs = [A(x) for x in range(20)]
    with ThreadPoolExecutor(max_workers=4) as executor:
        list(executor.map(lambda obj: obj.p(), runs))
    for run in runs:
        print(run.y)

内容的提问来源于stack exchange，提问作者feedMe