为什么我不能在类方法中使用 python 模块 concurrent.futures？答案

【问题标题】：Why I cannot use python module concurrent.futures in class method?为什么我不能在类方法中使用 python 模块 concurrent.futures？
【发布时间】：2013-06-29 11:36:31
【问题描述】：

我想让我的类方法并行运行，但它只会产生某种我无法解决的错误。我的代码是：

import concurrent.futures as futures

samples = ['asfd', 'zxcv', 'asf', 'qwer']

class test:
    def __init__(self, samples):
        maturedb = {}
        with futures.ProcessPoolExecutor() as exe:
            for samplename, dResult in exe.map(self.make_readdb, samples):
                maturedb[samplename] = dResult
        print(maturedb)

    def make_readdb(self, samplename):
        return samplename, 1

test(samples)

如果我在 Ubuntu 机器上运行此代码，则会出现如下错误：

Traceback (most recent call last):
    File "/usr/lib/python3.2/multiprocessing/queues.py", line 272, in _feedsend(obj)
    _pickle.PicklingError: Can't pickle <class 'method'>: attribute lookup builtins.method failed

方法make_readdb只是简化做个例子，但在实际代码中却是一个瓶颈，我需要让它平行。

【问题讨论】：

您的代码可以在 Python 3.3 上按原样运行
我在 Python 3.5.2 上测试了我的代码，它运行良好。非常感谢您的所有回答。

标签： python class parallel-processing class-method concurrent.futures

【解决方案1】：

来自docs：

ProcessPoolExecutor 类是一个使用池的 Executor 子类异步执行调用的进程。 ProcessPoolExecutor 使用多处理模块，它允许它绕过全局解释器锁也意味着只有可腌制的对象可以执行并返回。

试试ThreadPoolExecutor

我再次查看了您的代码，问题是函数 - make_readdb - 是类 test 的成员。你能重构并拉出这个函数吗？

【讨论】：

多线程根本没有帮助。由于 cPython 一次只运行一个线程，ThreadPoolExecutor 使它变得更慢。在这种情况下，单线程设计耗时约 600 秒，而 ThreadPoolExecutor 耗时约 1300 秒。
您的工作受 CPU 限制还是 IO 限制？你为 max_workers 使用了什么值？在您调用 ProcessPoolExecutor 时，这将默认为您机器中的处理器数量。

【解决方案2】：

self 应作为显式参数传递，即使在多个进程中也是如此。像这样：

class test:
    def __init__(self, samples):
        maturedb = {}
        with futures.ProcessPoolExecutor() as exe:
            for samplename, dResult in exe.map(test.make_readdb,self, samples):
                maturedb[samplename] = dResult
        print(maturedb)

    def make_readdb(self, samplename):
        return samplename, 1

但实际上只有一个进程会运行。所以这可能是一个更好的写法： Do not pass self to ProcessPoolExecutor in the class

class test:
    def __init__(self, samples):
        maturedb = {}
        with futures.ProcessPoolExecutor() as exe:
            for samplename, dResult in exe.map(test.make_readdb, samples):
                maturedb[samplename] = dResult
        print(maturedb)

    @staticmethod
    def make_readdb(samplename):
        return samplename, 1

【讨论】：