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

Question

我有这样的代码：

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

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

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

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

我不必使用concurrent，multiprocessing 也可以。这是一个类似的问题，不清楚如何在生成器中使用它。

轻微扭曲：生成器返回的每个值都是一个大型 numpy 数组（大约 10 兆字节）。我如何在没有酸洗和解酸的情况下转移它？我已经看过 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 数组，并且仍然受益于真正的并行性。