以下是使用
multiprocessing
并行化您的示例函数的方法。我还包括了一个几乎完全相同的纯 Python 函数,它使用非并行的
for
循环,以及一个可以实现相同结果的 numpy 一行代码:
import numpy as np
from multiprocessing import Pool
def f(x,y):
return x * y
def splat_f(args):
return f(*args)
pool = Pool(8)
def parallel(M, N):
results = pool.map(splat_f, ((i, j) for i in range(M) for j in range(N)))
return np.array(results).reshape(M, N)
def nonparallel(M, N):
out = np.zeros((M, N), np.int)
for i in range(M):
for j in range(N):
out[i, j] = f(i, j)
return out
def broadcast(M, N):
return np.prod(np.ogrid[:M, :N])
现在让我们来看一下性能:
%timeit parallel(1000, 1000)
# 1 loops, best of 3: 1.67 s per loop
%timeit nonparallel(1000, 1000)
# 1 loops, best of 3: 395 ms per loop
%timeit broadcast(1000, 1000)
# 100 loops, best of 3: 2 ms per loop
非并行的纯Python版本比并行版本快约4倍,而使用numpy数组广播的版本绝对压倒性优于其他两个版本。
问题在于启动和停止Python子进程会带来相当多的开销,并且您的测试函数如此微不足道,以至于每个工作线程只花费很少一部分的寿命来执行有用的工作。只有在每个线程被杀死之前有大量的工作要做时,多处理才有意义。例如,您可以给每个工作线程一个更大的输出数组块来计算(尝试调整pool.map()
的chunksize=
参数),但是基于这样微不足道的示例,我怀疑您不会看到很大的改进。
我不知道您实际的代码是什么样子的——也许您的函数足够大且昂贵,值得使用多处理。然而,我敢打赌,有更好的方法来提高其性能。