我该如何编写一个Python多进程脚本,使用以下两个队列?
1. 一个作为工作队列,它从一些数据开始,并根据要并行化的函数的条件,在运行时接收更多任务。 2. 另一个用于收集结果,并在处理完成后将结果写入其中。
基本上,我需要根据初始项中发现的内容向工作队列中添加更多任务。下面我发布的示例很愚蠢(我可以按照自己的意愿转换项目并直接放入输出队列),但其机制是清晰的,并反映了我需要开发的部分概念。这是我的尝试:
1. 一个作为工作队列,它从一些数据开始,并根据要并行化的函数的条件,在运行时接收更多任务。 2. 另一个用于收集结果,并在处理完成后将结果写入其中。
基本上,我需要根据初始项中发现的内容向工作队列中添加更多任务。下面我发布的示例很愚蠢(我可以按照自己的意愿转换项目并直接放入输出队列),但其机制是清晰的,并反映了我需要开发的部分概念。这是我的尝试:
import multiprocessing as mp
def worker(working_queue, output_queue):
item = working_queue.get() #I take an item from the working queue
if item % 2 == 0:
output_queue.put(item**2) # If I like it, I do something with it and conserve the result.
else:
working_queue.put(item+1) # If there is something missing, I do something with it and leave the result in the working queue
if __name__ == '__main__':
static_input = range(100)
working_q = mp.Queue()
output_q = mp.Queue()
for i in static_input:
working_q.put(i)
processes = [mp.Process(target=worker,args=(working_q, output_q)) for i in range(mp.cpu_count())] #I am running as many processes as CPU my machine has (is this wise?).
for proc in processes:
proc.start()
for proc in processes:
proc.join()
for result in iter(output_q.get, None):
print result #alternatively, I would like to (c)pickle.dump this, but I am not sure if it is possible.
这不会结束也不会打印任何结果。
在整个过程结束时,我希望确保工作队列为空,并且所有并行函数在后面迭代输出队列以取出结果之前都已经完成写入。你有什么建议可以使其正常工作吗?
try: print result except Empty: break
这将打印预期的总结果,但控制台输出仍在抱怨异常。我认为我没有正确处理它。 - Jaqo