使用池消除多处理中的开销

Question

我目前处于重复调用并行代码并尝试减少与多处理相关的开销的情况。因此，请考虑以下示例，该示例故意不包含“昂贵”的计算：

import multiprocessing as mp
def f(x):
    # toy function
    return x*x

if __name__ == '__main__':
    for x in range(500):
        pool = mp.Pool(processes=2) 
        print(pool.map(f, range(x, x + 50)))
        pool.close()
        pool.join()  # necessary?

此代码需要 53 秒，而顺序方法需要 0.04 秒。

第一个问题：在这种情况下，当只使用 pool.map() 时，我真的需要调用 pool.join() 吗？我找不到忽略它的任何负面影响，运行时间将下降到 4.8 秒。 （我知道省略 pool.close() 是不可能的，因为那时我们会泄漏线程。）

现在，虽然这将是一个很好的改进，但作为第一个答案，我可能会得到“好吧，首先不要在循环中创建池”。好的，没问题，但是并行化的代码实际上存在于实例方法中，所以我会使用：

class MyObject:
    def __init__(self):
        self.pool = mp.Pool(processes=2)
    def function(self, x):
        print(self.pool.map(f, range(x, x + 50)))

if __name__ == '__main__':
    my_object = MyObject()
    for x in range(500):
        my_object.function(x)

这将是我最喜欢的解决方案，因为它可以在 0.4 秒内运行。

第二个问题：我应该在某处显式调用 pool.close()/pool.join()（例如在 MyObject 的析构函数中）还是当前代码足够？ （如果重要的话：可以假设我的项目中只有几个长期存在的 MyObject 实例。）

Answer 1

当然这需要很长时间：你不断分配一个新池并为每个 x 销毁它。

如果你这样做，它会运行得更快：

if __name__ == '__main__':
    pool = mp.Pool(processes=2) # allocate the pool only once
    for x in range(500):
        print(pool.map(f, range(x, x + 50)))

    pool.close() # close it only after all the requests are submitted 
    pool.join() # wait for the last worker to finish

试试吧，你会发现它现在运行得更快了。

这里是join 和close 的文档链接：

一旦close 被调用，您就不能向池中提交更多任务，join 会一直等到最后一个工作人员完成其工作。应该按该顺序调用它们（先关闭然后加入）。

Answer 2

好吧，实际上您可以将已分配的池作为参数传递给您的对象：

class MyObject:
    def __init__(self, pool):
        self.pool = pool

    def function(self, x):
        print(self.pool.map(f, range(x, x + 50)))


if __name__ == '__main__':
    with mp.Pool(2) as pool:
        my_object = MyObject(pool)
        my_second_object = MyObject(pool)

        for x in range(500):
            my_object.function(x)
            my_second_object.function(x)

        pool.close()

我找不到可能需要在不同对象中使用不同池的原因