在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

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

out=(p.map([f1, f2], [x1, x2]))

对代码的最小更改是将 p.map() 调用替换为：

r1 = p.apply_async(f1, [x1])
out2 = f2(x2)
out1 = r1.get()

尽管如果您只想同时运行两个函数调用，那么您不需要这里的Pool()，您可以手动启动一个线程/进程和use Pipe/Queue to get the result：

#!/usr/bin/env python
from multiprocessing import Process, Pipe

def another_process(f, args, conn):
    conn.send(f(*args))
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe(duplex=False)
    p = Process(target=another_process, args=(f1, [x1], child_conn))
    p.start()
    out2 = f2(x2)
    out1 = parent_conn.recv()
    p.join()

Answer 2

我相信您想在代码中使用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 对象，例如一个函数。让我们看看该方法的文档。

对iterable中的每个元素应用func，将结果收集在一个 返回的列表。

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

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