将多个参数传递给 concurrent.futures.Executor.map？

Question

concurrent.futures.Executor.map 采用可变数量的可迭代对象，从中调用给定的函数。 如果我有一个生成元组的生成器通常在原地解包，我应该怎么称呼它？

以下内容不起作用，因为每个生成的元组都作为 map 的不同参数给出：

args = ((a, b) for (a, b) in c)
for result in executor.map(f, *args):
    pass

如果没有生成器，映射所需的参数可能如下所示：

executor.map(
    f,
    (i[0] for i in args),
    (i[1] for i in args),
    ...,
    (i[N] for i in args),
)

Answer 1

一个参数重复，c中的一个参数

from itertools import repeat
for result in executor.map(f, repeat(a), c):
    pass

需要解包c的物品，可以解包c

from itertools import izip
for result in executor.map(f, *izip(*c)):
    pass

需要解压c的物品，无法解压c

1. 将f 更改为采用单个参数并在函数中解压缩参数。

2. 如果c 中的每个项目都有可变数量的成员，或者您只调用了几次f：

   executor.map(lambda args, f=f: f(*args), c)
   

   它定义了一个新函数，从c 解包每个项目并调用f。在lambda 中使用f 的默认参数会使f 在lambda 中成为局部变量，从而减少查找时间。

3. 如果您有固定数量的参数，并且需要多次调用f：

   from collections import deque
   def itemtee(iterable, n=2):
       def gen(it = iter(iterable), items = deque(), next = next):
           popleft = items.popleft
           extend = items.extend
           while True:
               if not items:
                   extend(next(it))
               yield popleft()
       return [gen()] * n
   
   executor.map(f, *itemtee(c, n))
   

其中n 是f 的参数数量。本文改编自itertools.tee。

Answer 2

您需要在map 调用中删除*：

args = ((a, b) for b in c)
for result in executor.map(f, args):
    pass

这将调用f、len(args) 次，其中f 应该接受一个参数。

如果您希望f 接受两个参数，您可以使用如下 lambda 调用：

args = ((a, b) for b in c)
for result in executor.map(lambda p: f(*p), args):   # (*p) does the unpacking part
    pass

Answer 3

你可以使用currying在Python中通过partial方法创建新函数

from concurrent.futures import ThreadPoolExecutor
from functools import partial


def some_func(param1, param2):
    # some code

# currying some_func with 'a' argument is repeated
func = partial(some_func, a)
with ThreadPoolExecutor() as executor:
    executor.map(func, list_of_args):
    ...

如果您需要传递多个相同的参数，您可以将它们传递给partial 方法

func = partial(some_func, a, b, c)

Answer 4

所以假设你有一个函数，它接受 3 个参数，并且所有 3 个参数都是 动态 并随着每次调用而不断变化。例如：

def multiply(a,b,c):
    print(a * b * c)

要使用线程多次调用它，我首先会创建一个元组列表，其中每个元组都是 a、b、c 的一个版本：

arguments = [(1,2,3), (4,5,6), (7,8,9), ....]

据我们所知，concurrent.futures 的map 函数将接受第一个参数作为目标函数，第二个参数作为每个版本的参数列表将要执行的函数。因此，您可能会这样调用：

for _ in executor.map(multiply, arguments) # Error

但这会给你错误该函数预期3 arguments but got only 1。为了解决这个问题，我们创建了一个辅助函数：

def helper(numbers):
    multiply(numbers[0], numbers[1], numbers[2])

现在，我们可以使用 executor 调用这个函数，如下所示：

with ThreadPoolExecutor() as executor:
     for _ in executor.map(helper, arguments):
         pass

这应该会给你想要的结果。

Answer 5

这是一段代码 sn-p，展示了如何使用 ThreadPoolExecutor 向函数发送多个参数：

import concurrent.futures


def hello(first_name: str, last_name: str) -> None:
    """Prints a friendly hello with first name and last name"""
    print('Hello %s %s!' % (first_name, last_name))


def main() -> None:
    """Examples showing how to use ThreadPoolExecutor and executer.map
    sending multiple arguments to a function"""

    # Example 1: Sending multiple arguments using tuples
    # Define tuples with sequential arguments to be passed to hello()
    args_names = (
        ('Bruce', 'Wayne'),
        ('Clark', 'Kent'),
        ('Diana', 'Prince'),
        ('Barry', 'Allen'),
    )
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Using lambda, unpacks the tuple (*f) into hello(*args)
        executor.map(lambda f: hello(*f), args_names)

    print()

    # Example 2: Sending multiple arguments using dict with named keys
    # Define dicts with arguments as key names to be passed to hello()
    kwargs_names = (
        {'first_name': 'Bruce', 'last_name': 'Wayne'},
        {'first_name': 'Clark', 'last_name': 'Kent'},
        {'first_name': 'Diana', 'last_name': 'Prince'},
        {'first_name': 'Barry', 'last_name': 'Allen'},
    )
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Using lambda, unpacks the dict (**f) into hello(**kwargs)
        executor.map(lambda f: hello(**f), kwargs_names)


if __name__ == '__main__':
    main()

Answer 6

对于ProcessPoolExecutor.map()：

类似于 map(func, *iterables) 除了：

iterables 被立即收集而不是延迟收集；

func 是异步执行的，可以多次调用 func 同时。

尝试在python 3下运行下面的sn-p，你就会很清楚了：

from concurrent.futures import ProcessPoolExecutor

def f(a, b):
    print(a+b)

with ProcessPoolExecutor() as pool:
    pool.map(f, (0, 1, 2, 3, 4, 5, 6, 7, 8, 9), (0, 1, 2))

# 0, 2, 4

array = [(i, i) for i in range(3)]
with ProcessPoolExecutor() as pool:
    pool.map(f, *zip(*array))

# 0, 2, 4

Answer 7

我在这里看到了很多答案，但没有一个比使用 lambda 表达式更直接：

foo(x,y): 通过

想以相同的值调用上述方法 10 次，即 xVal 和 yVal？ 以 concurrent.futures.ThreadPoolExecutor() 作为执行者：

for _ in executor.map( lambda _: foo(xVal, yVal), range(0, 10)):
    pass

Answer 8

假设您在下面显示的数据框中有这样的数据，并且您想将第一两列传递给一个函数，该函数将读取图像并预测特征，然后计算差异并返回差异值。

注意：您可以根据您的要求有任何场景，您可以分别定义功能。

下面的代码 sn-p 会将这两列作为参数传递给线程池机制（同时显示进度条）

''' function that will give the difference of two numpy feature matrix'''
def getDifference(image_1_loc, image_2_loc, esp=1e-7):
       arr1 = ''' read 1st image and extract feature '''
       arr2 = ''' read 2nd image and extract feature '''
       diff = arr1.ravel() - arr2.ravel() + esp    
       return diff

'''Using ThreadPoolExecutor from concurrent.futures with multiple argument'''

with ThreadPoolExecutor() as executor:
        result = np.array(
                         list(tqdm(
                                   executor.map(lambda x : function(*x), [(i,j) for i,j in df[['image_1','image_2']].values]),
                               total=len(df)
                                  ) 
                             )
                          )