Python Multiprocessing apply_async 不可腌制？

Question

我正在计算大量函数（大约 1000000），由于它非常耗时，我正在使用 multiprocessing.Pool.apply_async 函数。但是，当我尝试使用 AsyncResult 类的 .get() 函数读取结果时，出现错误：

File "Test.py", line 17, in <module>
    Test()
  File "Test.py", line 11, in __init__
    self.testList[i].get(5)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 657, in get
    raise self._value
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 431, in _handle_tasks
    put(task)
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

给出相同错误的简化类：

import multiprocessing as mp
import numpy as np

class Test:
    def __init__(self):
        pool = mp.Pool(processes = 4)
        self.testList = [0,0,0,0]
        for i in range(0,len(self.testList)):
            self.testList[i] = pool.apply_async(self.run, (1,))
        for i in range(0,len(self.testList)):
            self.testList[i].get(5)

    def run(self, i):
        return 1


Test()

有趣的是，如果我改为使用 self.testList testList，则代码可以正常工作。但是，当我使用 .ready() 而不是 .get() 比较两者时，我发现 self.testList 比 testList 快大约 1000 倍（我无法解释）。所以，我真的很想找到一种使用 self.testList 的方法。

我一直在四处寻找，尽管还有其他关于此的线程，但它们似乎更关注队列而不是 apply_async。任何帮助将不胜感激！

谢谢！

编辑：看起来最初的问题是因为我在一个类中调用 mp.Pool 。当我在类之外创建相同的进程时，程序会运行，但与类中的代码相比，它非常慢（慢 30 倍）（我使用 .ready() 函数对此进行了测试，在这两种情况下都可以正常工作） .这是一个最小的例子：

import multiprocessing as mp
import numpy as np
import time

class Test:
    def __init__(self):
        pool = mp.Pool(processes = 4)
        self.testList = [0 for i in range(0,100000)]
        for i in range(0,len(self.testList)):
            self.testList[i] = pool.apply_async(self.run, (1,))
        for i in range(0,len(self.testList)):
            while not self.testList[i].ready():
                continue

    def run(self, i):
        return 1

def functionTest():
    pool = mp.Pool(processes = 4)
    testList = [0 for i in range(0,100000)]
    for i in range(0,len(testList)):
        testList[i] = pool.apply_async(run, (1,))
    for i in range(0,len(testList)):
        while not testList[i].ready():
            continue

def run(i):
    return 1


startTime1 = time.time()
Test()
startTime2 = time.time()
print(startTime2-startTime1)



startTime1 = time.time()
functionTest()
startTime2 = time.time()
print(startTime2-startTime1)

这个测试的输出是

5.861901044845581
151.7218940258026

我尝试寻找使类方法发挥作用的方法，例如将多处理从 init 函数中取出或将池对象提供给类而不是让类创建它。不幸的是，这些方法都不起作用。我真的很想找到一种可行且速度仍然很快的方法。感谢您的帮助！

Answer 1

当您产生多个线程时，您正试图腌制整个类，其中包含来自 init 中设置的mp.Pool 的值。复制 mp.Pool 两者都不起作用，在这里也没有任何意义。而是将您的类拆分为两个单独的顶级函数，或者至少将多处理内容移到它自己的函数中，在 Test 类之外。