Python 多处理 apply_async 不可pickle？

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

当您生成多个线程时，您试图对整个类进行 pickle，其中包含 init 中设置的 mp.Pool 中的值。复制 mp.Pool 既不起作用，也没有真正的意义。将您的类拆分为两个单独的顶级函数，或者至少将多处理内容移至 Test 类之外的其自己的函数中。