Как написать многопоточную функцию для одновременной обработки разных задач?

Question

Я хотел бы определить функцию do_in_parallel в python, которая будет принимать функции с аргументами, создавать потоки для каждого и выполнять их параллельно.Функция должна работать следующим образом:

do_in_parallel(_sleep(3), _sleep(8), _sleep(3))

Однако мне трудно определить функцию do_in_parallel, которая бы принимала несколько функций с несколькими аргументами в каждой, вот моя попытка:

from time import sleep
import threading

def do_in_parallel(*kwargs):

    tasks = []

    for func in kwargs.keys():
        t = threading.Thread(target=func, args=(arg for arg in kwargs[func]))
        t.start()
        tasks.append(t)

    for task in tasks:        
        task.join()

def _sleep(n):
    sleep(n)
    print('slept', n)

Используя его так, и получая следующую ошибку:

do_in_parallel(_sleep=3, _sleep=8, _sleep=3)

>> do_in_parallel(sleepX=3, sleepX=8, sleepX=3)
                            ^
>> SyntaxError: keyword argument repeated

Может кто-нибудь объяснить, что мне нужно изменить в моей функции, чтобы она могла принимать несколько параметров функции следующим образом:

do_in_parallel(_sleep(3), _sleep(8), maybe_do_something(else, and_else))

Answer 1

do_in_parallel(_sleep(3), _sleep(8), maybe_do_something(else, and_else))

Эта структура вызова не будет работать в любом случае, поскольку вы передаете результаты своих целевых функций в do_in_parallel (вы уже вызываете _sleep и т. Д.).

Вместо этого вам нужно связать задач и передать эти задачи вашей функции обработки.Задачей здесь является кортеж, содержащий целевую функцию, которую нужно вызвать, и аргумент-кортеж task = (_sleep, (n,)).

. Затем я предлагаю вам использовать ThreadPool и метод apply_async для обработки отдельных задач.

from time import sleep
from multiprocessing.dummy import Pool  # .dummy.Pool is a ThreadPool


def _sleep(n):
    sleep(n)
    result = f'slept {n}'
    print(result)
    return result


def _add(a, b):
    result = a + b
    print(result)
    return result


def do_threaded(tasks):
    with Pool(len(tasks)) as pool:
        results = [pool.apply_async(*t) for t in tasks]
        results = [res.get() for res in results]
    return results


if __name__ == '__main__':

    tasks = [(_sleep, (i,)) for i in [3, 8, 3]]
    # [(<function _sleep at 0x7f035f844ea0>, (3,)),
    #  (<function _sleep at 0x7f035f844ea0>, (8,)),
    #  (<function _sleep at 0x7f035f844ea0>, (3,))]
    tasks += [(_add, (a, b)) for a, b in zip(range(0, 3), range(10, 13))]

    print(do_threaded(tasks))

Вывод:

10
12
14
slept 3
slept 3
slept 8
['slept 3', 'slept 8', 'slept 3', 10, 12, 14]

Process finished with exit code 0