如何并行运行生成器代码？

Question

我有这样的代码：

def generator():
    while True:
        # do slow calculation
        yield x

我想将缓慢的计算转移到单独的进程中。

我正在 python 3.6 中工作，所以我有concurrent.futures.ProcessPoolExecutor. 如何使用它来并发生成器并不明显。

与常规并发场景使用的区别map在于，这里没有什么可映射的（生成器永远运行），并且我们不希望一次获得所有结果，我们希望将它们排队并等到队列未满之前计算更多结果。

我不用用concurrent，multiprocessing也可以。这是一个类似的问题，如何在生成器内部使用它并不明显。

略有不同：生成器返回的每个值都是一个大的 numpy 数组（10 MB 左右）。如何在不酸洗和不酸洗的情况下转移它？我已经看过文档，multiprocessing.Array但如何使用它来传输 numpy 数组并不完全明显。

Answer 1

在这种情况下，我通常使用joblib库。它是一个基于多处理的并行计算框架。对于必须处理大型 numpy 数组的情况，它精确地支持内存映射。我相信值得您检查一下。

也许 joblib 的文档在这一点上不够明确，只显示了 for 循环的示例，因为您想使用生成器，我应该指出它确实可以与生成器一起使用。可以实现您想要的效果的示例如下：

from joblib import Parallel, delayed
def my_long_running_job(x):
    # do something with x
# you can customize the number of jobs
Parallel(n_jobs=4)(delayed(my_long_running_job)(x) for x in generator())

编辑：我不知道你想要做什么类型的处理，但如果它释放了 GIL，你也可以考虑使用线程。这样，您就不会遇到必须在进程之间传输大型 numpy 数组的问题，并且仍然受益于真正的并行性。