python - 如何在 Python 中并行化生成器/迭代器的管道？

Question

假设我有如下一些 Python 代码:

input = open("input.txt")
x = (process_line(line) for line in input)
y = (process_item(item) for item in x)
z = (generate_output_line(item) + "\n" for item in y)
output = open("output.txt", "w")
output.writelines(z)

此代码从输入文件中读取每一行，通过几个函数运行它，并将输出写入输出文件。现在我知道函数process_line、process_item 和generate_output_line 永远不会相互干扰，让我们假设输入和输出文件在不同的磁盘上，这样读写就不会互相干扰。

但是 Python 可能不知道这些。我的理解是 Python 将读取一行，依次应用每个函数，并将结果写入输出，然后它只会在将第一行发送到输出后读取第二行，所以在第一条线退出之前，第二条线不会进入管道。我是否正确理解了该程序的流程？如果这是它的工作原理，是否有任何简单的方法可以使多条线同时进入管道，以便程序并行读取、写入和处理每个步骤？

Answer 1

你不能真正并行读取或写入文件；这些最终将成为你的瓶颈。您确定这里的瓶颈是 CPU，而不是 I/O？

由于您的处理不包含依赖项(根据您的说法)，因此使用起来非常简单 Python's multiprocessing.Pool class .

有几种写法，但是 w.r.t. 更简单。调试是为了找到独立的关键路径(代码中最慢的部分)，我们将使其并行运行。我们假设它是 process_item。

……实际上就是这样。代码:

import multiprocessing.Pool

p = multiprocessing.Pool() # use all available CPUs

input = open("input.txt")
x = (process_line(line) for line in input)
y = p.imap(process_item, x)
z = (generate_output_line(item) + "\n" for item in y)
output = open("output.txt", "w")
output.writelines(z)

我还没有测试过，但这是基本思路。 Pool 的 imap 方法确保以正确的顺序返回结果。