如何解压来自 ProcessPoolExecutor 的生成器结果？

Question

我正在尝试使用 ProcessPoolExecutor 来启动一个 class (BT) 的多个实例，它启动一个方法 BT.bt()，并且 return结果 [aList1, aList2].

我试图将 class 的每个实例所需的信息打包到一个简单的迭代器 (bulkInfo) 中。然后将迭代器抛给 class 的每个实例并执行。据我了解，结果应该是生成器或列表，结果中的每个结果（combiList）都是 2 'minilists' [aList1, aList2] 的列表.

但是我无法解压结果。或者更准确地说，我无法解压每个 result in results。 Pycharm 突出显示 result[0] 和 result[1] 处的警告 — Class 'BT' does not define '__getitem__', so the [] operator cannot be used on its instances.

运行代码给出错误 TypeError: 'list' object is not callable。完整版代码报错TypeError: cannot pickle 'generator' object.

如何解压results中的每个result给我miniList1和miniList2？

下面的代码是一个精简版本，用于概述问题。我试过使用 yield 而不是 return combiList。我试过将 results 扔到 list(results) 并遍历每个结果。我尝试了多种将 bulkinfo 迭代器打包为列表列表或元组列表的变体。我尝试将 combiList 定义为列表和元组。我已经尝试 list(results) 然后 list(result) 来访问每个 miniList.

import concurrent.futures as cf
import numpy as np

class BT:
    def __init__(self, bulkInfo):
        self.aCyc = bulkInfo(3)
        self.bt(self.aCyc)

    def bt(self, aCycle):
        aList1 = []
        aList2 = []
        someInfo = [aCycle, 'bunch', 'of', 'stuff']
        [aList1.append(s) for s in someInfo]
        [aList2.append(s) for s in someInfo]
        combiList = [aList1, aList2]
        return combiList

if __name__ == '__main__':
    dummy1 = 'something'
    cycles = np.arange(10)
    bulkInfo = []
    [bulkInfo.append([dummy1, dummy1, dummy1, aCyc]) for aCyc in cycles]

    with cf.ProcessPoolExecutor(max_workers=4) as executor:
        results = executor.map(BT, bulkInfo)

        for result in results:
            miniList1 = result[0]
            miniList2 = result[1]

        # for result in list(results):
        #     for miniList in list(result):
        #         print(miniList)

我不太关心性能，只是想从 ProcessPoolExecutor 中获取结果。所以我只是想将输出结果作为一个列表，但也许它是一个生成器（当列表“成为”生成器时，我在这里不清楚，反之亦然。我只想提取 miniList1 和 miniList2.)

Answer 1

您的 BT 对象存在一些问题。

首先，在__init__()中，你不能用()访问bulkInfo的元素（这是一个列表），你需要使用括号。这就是你得到 TypeError: 'list' object is not callable 的原因。这是更正后的版本：

def __init__(self, bulkInfo):
    self.aCyc = bulkInfo[3]
    self.bt(self.aCyc)

其次，您的 init 方法 returns 一个 BT 对象，而不是列表的列表。这就是 __init__ 方法的作用：它创建一个对象并 returns 它。因此，executor.map 的内容是 BT 对象的可迭代对象。通过将 combiList 更改为对象的属性，您可以使用 result.combiList[i].

访问它

虽然我们在这里，但您不必将 self.aCyc 作为 bt() 方法的参数传递：因为它是一个属性，您可以在对象方法中自由访问它.

最后，您还可以直接遍历 executor.map 而无需 results 变量。

这是完整的工作代码：

import concurrent.futures as cf
import numpy as np

class BT:
    def __init__(self, bulkInfo):
        self.aCyc = bulkInfo[3]
        self.bt()

    def bt(self):
        aList1 = []
        aList2 = []
        someInfo = [self.aCyc, 'bunch', 'of', 'stuff']
        [aList1.append(s) for s in someInfo]
        [aList2.append(s) for s in someInfo]
        self.combiList = [aList1, aList2]

if __name__ == '__main__':
    dummy1 = 'something'
    cycles = np.arange(10)
    bulkInfo = []
    [bulkInfo.append([dummy1, dummy1, dummy1, aCyc]) for aCyc in cycles]

    with cf.ProcessPoolExecutor(max_workers=4) as executor:
        for result in executor.map(BT, bulkInfo):
            print(result.combiList)
            miniList1 = result.combiList[0]
            miniList2 = result.combiList[1]

这给了我以下输出：

[[0, 'bunch', 'of', 'stuff'], [0, 'bunch', 'of', 'stuff']]
[[1, 'bunch', 'of', 'stuff'], [1, 'bunch', 'of', 'stuff']]
[[2, 'bunch', 'of', 'stuff'], [2, 'bunch', 'of', 'stuff']]
[[3, 'bunch', 'of', 'stuff'], [3, 'bunch', 'of', 'stuff']]
[[4, 'bunch', 'of', 'stuff'], [4, 'bunch', 'of', 'stuff']]
[[5, 'bunch', 'of', 'stuff'], [5, 'bunch', 'of', 'stuff']]
[[6, 'bunch', 'of', 'stuff'], [6, 'bunch', 'of', 'stuff']]
[[7, 'bunch', 'of', 'stuff'], [7, 'bunch', 'of', 'stuff']]
[[8, 'bunch', 'of', 'stuff'], [8, 'bunch', 'of', 'stuff']]
[[9, 'bunch', 'of', 'stuff'], [9, 'bunch', 'of', 'stuff']]

Answer 2

        with concurrent.futures.ProcessPoolExecutor() as executor:
        results = executor.map(self._translate, itertools.repeat(text), languages)
        rlist = []
        for res in results:
            rlist.append(res)
            # print(type(res))
        return rlist

您可以像这里一样创建列表并附加它rlist.append(res)

如何解压来自 ProcessPoolExecutor 的生成器结果？

How to unpack generator results from ProcessPoolExecutor?

python

numpy

multiprocessing

generator

concurrent.futures