我在一个类中有一个方法，需要在循环中完成大量工作，我想将工作分散到我的所有核心上。

我写了下面的代码，如果我使用正常的话，它可以工作map()，但与pool.map()返回错误。

import multiprocessing
pool = multiprocessing.Pool(multiprocessing.cpu_count() - 1)

class OtherClass:
  def run(sentence, graph):
    return False

class SomeClass:
  def __init__(self):
    self.sentences = [["Some string"]]
    self.graphs = ["string"]

  def some_method(self):
      other = OtherClass()

      def single(params):
          sentences, graph = params
          return [other.run(sentence, graph) for sentence in sentences]

      return list(pool.map(single, zip(self.sentences, self.graphs)))


SomeClass().some_method()

Error 1:

AttributeError：无法腌制本地对象“SomeClass.some_method..single”

为什么不能腌制single()？我什至尝试移动single()到全局模块范围（不在类内部 - 使其独立于上下文）：

import multiprocessing
pool = multiprocessing.Pool(multiprocessing.cpu_count() - 1)

class OtherClass:
  def run(sentence, graph):
    return False


def single(params):
    other = OtherClass()
    sentences, graph = params
    return [other.run(sentence, graph) for sentence in sentences]

class SomeClass:
  def __init__(self):
    self.sentences = [["Some string"]]
    self.graphs = ["string"]

  def some_method(self):
      return list(pool.map(single, zip(self.sentences, self.graphs)))


SomeClass().some_method()

我得到以下结果...

Error 2:

AttributeError：无法在模块'上获取属性'single'main' 来自 '.../test.py'

Error 1:

属性错误：无法腌制本地对象 'SomeClass.some_method..single'

您通过移动嵌套目标函数自行解决了此错误single()到顶层。

背景：

池需要腌制（序列化）它发送到其工作进程的所有内容（IPC https://en.wikipedia.org/wiki/Inter-process_communication）。 Pickling 实际上只保存函数的名称，而 unpickle 则需要按名称重新导入函数。为了实现这一点，该函数需要在顶层定义，嵌套函数将无法被子函数导入，并且已经尝试腌制它们会引发异常（more https://stackoverflow.com/a/56534386/9059420).

Error 2:

AttributeError：无法从模块“main”获取属性“single” '.../test.py'

您正在启动池before您定义函数和类，这样子进程就无法继承任何代码。将泳池启动位置移至底部并保护（why? https://stackoverflow.com/a/52693952/9059420）它与if __name__ == '__main__':

import multiprocessing

class OtherClass:
  def run(self, sentence, graph):
    return False


def single(params):
    other = OtherClass()
    sentences, graph = params
    return [other.run(sentence, graph) for sentence in sentences]

class SomeClass:
   def __init__(self):
       self.sentences = [["Some string"]]
       self.graphs = ["string"]

   def some_method(self):
      return list(pool.map(single, zip(self.sentences, self.graphs)))

if __name__ == '__main__':  # <- prevent RuntimeError for 'spawn'
    # and 'forkserver' start_methods
    with multiprocessing.Pool(multiprocessing.cpu_count() - 1) as pool:
        print(SomeClass().some_method())

Appendix

...我想将工作分散到我的所有核心。

可能有用的背景知识multiprocessing.Pool是分块工作：

Python 多处理：理解 chunksize 背后的逻辑 https://stackoverflow.com/q/53751050/9059420