为啥我不能在类方法中使用 python 模块 concurrent.futures？

Posted 2023-02-16

【中文标题】为啥我不能在类方法中使用 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 上测试了我的代码，它运行良好。非常感谢您的所有回答。

【参考方案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