如何同时从 multiprocessing.Queue() 中 put() 和 get()？

Question

我正在开发一个python 2.7 程序，它使用multiprocessing 并行执行这些操作：

• 同时从file 1和file 2读取一行
• 申请function(line_1, line_2)
• 将函数输出写入文件

我是多处理的新手，一般来说我对 python 并不是非常熟悉。因此，我阅读了很多已经提出的问题和教程：我觉得很接近重点，但我现在可能遗漏了一些我无法真正发现的东西。

代码结构如下：

from itertools import izip
from multiprocessing import Queue, Process, Lock

nthreads = int(mp.cpu_count())
outq = Queue(nthreads)
l = Lock()

def func(record_1, record_2):
     result = # do stuff
     outq.put(result)

OUT = open("outputfile.txt", "w")
IN1 = open("infile_1.txt", "r")
IN2 = open("infile_2.txt", "r")

processes = []
for record_1, record_2 in izip(IN1, IN2):
     proc = Process(target=func, args=(record_1, record_2))
     processes.append(proc)
     proc.start()

for proc in processes:
     proc.join()

while (not outq.empty()):
     l.acquire()
     item = outq.get()
     OUT.write(item)
     l.release()

OUT.close()
IN1.close()
IN2.close()

据我了解（到目前为止）作为包的多处理，我正在做的是：

• 为函数的结果创建队列，该队列的大小限制与机器的内核数兼容。
• 用func() 的结果填充此队列。
• 读取队列项直到队列为空，然后将它们写入输出文件。

现在，我的问题是，当我运行这个脚本时，它立即变成了一个僵尸进程。我知道该功能有效，因为没有多处理实现，我得到了我想要的结果。

我想同时读取这两个文件并写入输出，以避免从我的输入文件生成一个巨大的列表然后读取它（输入文件很大）。您是否看到任何严重、完全错误或可改进的地方？

Answer 1

我看到的最大问题是您应该通过进程传递队列对象，而不是尝试在函数中将其用作全局对象。

def func(record_1, record_2, queue):
     result = # do stuff
     queue.put(result)

for record_1, record_2 in izip(IN1, IN2):
     proc = Process(target=func, args=(record_1, record_2, outq))

此外，正如目前所写的那样，您仍然会将所有这些信息拉入内存（也称为队列）并等待读取完成，然后再写入输出文件。您需要移动 p.join 循环直到读完队列之后，而不是将所有信息放在队列中 func 的末尾，它应该随着时间的推移在循环中用卡盘填充队列，否则这与将其全部读入内存相同。

你也不需要锁，除非你在工作函数func中使用它，如果你这样做了，你会再次想要通过它。

如果您不想在内存中读取/存储大量内容，我会在遍历输入文件的同时写出。这是将文件的每一行组合在一起的基本示例。

with open("infile_1.txt") as infile1, open("infile_2.txt") as infile2, open("out", "w") as outfile:
    for line1, line2 in zip(infile1, infile2):
        outfile.write(line1 + line2)

我不想写太多关于所有这些的东西，只是想给你一些想法。如果您想了解有关某事的更多详细信息，请告诉我。希望对您有所帮助！