python - 使用部分方法将多个可迭代参数传递给 ProcessPoolExecutor.map

Question

设置

我设置了一个函数来接收多个关键字参数:

def process(image, folder, param1, param2, param3):
    do_things
    return

我有几个 param1/2/3 的组合，我想以完全排列的方式遍历它们(即 1,4,7 然后 1,4,8 直到 3,6,9):

param1 = [1,2,3]
param2 = [4,5,6]
param3 = [7,8,9]

但是，image 和 folder 保持不变，所以我设置了一个部分，并将其包装在 ProcessPoolExecutor 映射函数中:

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        ...
    )

问题

当我有 1 个函数参数，并从 process() 中删除 param2 和 param3 时，以下单个可迭代工作:

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [q for q in param1]
    )

但是我如何为 executor.map 传递多个可迭代对象呢？

我试过了

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [q for q in param1],
        [q for q in param2],
        [q for q in param3],
    )

无论是否指定函数关键字名称，但无论哪种情况，我都会得到一个 TypeError: process() got multiple values for argument 'image'

还试过:

    with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(process,
            image=image,
            folder=folder,
        ),
        [i for i in product(param1, param2, param3)]
    )

上面给出了错误:BrokenProcessPool:进程池中的进程在未来运行或挂起时突然终止。

预期结果

我希望 executor.map 遍历多个可迭代对象并将它们传递给正确的函数关键字。

可重现错误代码:

from functools import partial
from concurrent import futures
from itertools import product

def new_fn(a, b, c):
    print(a, b, c)
    return (a, b, c)

a_static = 'Hi: '
b_it = ['b1', 'b2', 'b3']
c_it = ['c1', 'c2', 'c3']
p = [i for i in product(b_it, c_it)]

with futures.ProcessPoolExecutor(max_workers=4) as executor:
    data = executor.map(
        partial(new_fn, a=a_static), p
    )

data

Answer 1

您永远不会确切地说您想如何遍历 param1/2/3，所以这只是一个猜测。

您可以通过稍微更改 process() 函数的调用顺序并使用内置的 zip() 来做我想做的事。 unzip 数据的功能(如文档中所述)。

请注意内置的 partial() 函数是如何被使用的——正如您所尝试的那样，尝试将位置参数的关键字参数传递给它是不正确的。

我还添加了 if __name__ == '__main__': 使 Python 多处理代码(也)在 Windows 上运行所需的守卫。

import concurrent.futures as futures
from functools import partial


def process(image, folder, params):
    param1, param2, param3 = params
    print(f'param1={param1}, param2={param2}, param3={param3}')


if __name__ == '__main__':

    image = 'image'
    folder = 'folder'

    param1 = [1, 2, 3]
    param2 = [4, 5, 6]
    param3 = [7, 8, 9]

    with futures.ProcessPoolExecutor(max_workers=4) as executor:
        data = executor.map(
            partial(process, image, folder),
            zip(*zip(param1, param2, param3))
        )

        for _ in data:
            ...

输出:

param1=1, param2=2, param3=3
param1=4, param2=5, param3=6
param1=7, param2=8, param3=9