python - 在python中同时并行化不同的功能

Question

我想同时执行f1和f2。但是下面的代码不起作用!

from multiprocessing import Pool

def f1(x):
return x*x

def f2(x):
return x^2

if __name__ == '__main__':

    x1=10
    x2=20
    p= Pool(2)
    out=(p.map([f1, f2], [x1, x2]))

y1=out[0]
y2=out[1]

Answer 1

我相信你会喜欢使用 threading.Thread和 shared queue在您的代码中。

from queue import Queue
from threading import Thread
import time

def f1(q, x):
    # Sleep function added to compare execution times.
    time.sleep(5)
    # Instead of returning the result we put it in shared queue.
    q.put(x * 2)

def f2(q, x):
    time.sleep(5)
    q.put(x ^ 2)

if __name__ == '__main__':
    x1 = 10
    x2 = 20
    result_queue = Queue()

    # We create two threads and pass shared queue to both of them.
    t1 = Thread(target=f1, args=(result_queue, x1))
    t2 = Thread(target=f2, args=(result_queue, x2))

    # Starting threads...
    print("Start: %s" % time.ctime())
    t1.start()
    t2.start()

    # Waiting for threads to finish execution...
    t1.join()
    t2.join()
    print("End:   %s" % time.ctime())

    # After threads are done, we can read results from the queue.
    while not result_queue.empty():
        result = result_queue.get()
        print(result)

上面的代码应该输出类似于:

Start: Sat Jul  2 20:50:50 2016
End:   Sat Jul  2 20:50:55 2016
20
22

如您所见，即使两个函数都等待 5 秒才能产生结果，但它们是并行执行的，因此总执行时间为 5 秒。

如果您关心什么函数将什么结果放入您的队列，我可以看到两个解决方案可以确定这一点。您可以创建多个队列或将结果包装在一个元组中。

def f1(q, x):
    time.sleep(5)
    # Tuple containing function information.
    q.put((f1, x * 2))

并且为了进一步简化(尤其是当您有许多函数要处理时)，您可以装饰您的函数(以避免重复代码并允许无队列的函数调用):

def wrap_result(func):
    def wrapper(*args):
        # Assuming that shared queue is always the last argument.
        q = args[len(args) - 1]
        # We use it to store the results only if it was provided.
        if isinstance(q, Queue):
            function_result = func(*args[:-1])
            q.put((func, function_result))
        else:
            function_result = func(*args)
        return function_result

    return wrapper

@wrap_result
def f1(x):
    time.sleep(5)
    return x * 2

请注意，我的装饰器是匆忙编写的，它的实现可能需要改进(例如，如果您的函数接受 kwargs)。如果您决定使用它，则必须以相反的顺序传递参数:t1 = threading.Thread(target=f1, args=(x1, result_queue)) .

一点友善的建议。

“以下代码不起作用”并未说明该问题。是否引发异常？是否给出了意想不到的结果？

阅读错误信息很重要。更重要的是 - 研究它们的含义。您提供的代码会引发 TypeError带有非常明显的信息:

File ".../stack.py", line 16, in <module> out = (p.map([f1, f2], [x1, x2]))

TypeError: 'list' object is not callable

这意味着 Pool().map() 的第一个参数必须是 callable对象，例如一个函数。让我们看看该方法的文档。

Apply func to each element in iterable, collecting the results in a list that is returned.

它显然不允许将函数列表作为参数传递。

Here你可以阅读更多关于 Pool().map() 的信息方法。