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

Question

在concurrent.futures.Executor.map采用可变数目从该给出的函数被调用iterables的.如果我有一个生成通常解压缩的元组的生成器,我应该如何调用它？

以下方法不起作用,因为每个生成的元组都作为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.f在lambdamake flocal中使用默认参数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

下面的代码片段展示了如何使用 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 4

您可以使用currying通过Python中的部分方法创建新函数

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):
    ...

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

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

Answer 5

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

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

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

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

我们知道 thatconcurrent.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

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