Python 多处理 - 'Queue' 对象没有属性 'task_done'/'join'

Question

我正在将线程进程重写为多处理队列，以尝试加速大型计算。我已经完成了 95% 的进度，但我不知道如何在 Queue 时发出信号使用 multiprocessing 为空.

---

我的原始代码是这样的:

import Queue
from threading import Thread

num_fetch_threads = 4
enclosure_queue = Queue()

for i in range(num_fetch_threads):
  worker = Thread(target=run_experiment, args=(i, enclosure_queue))
  worker.setDaemon(True)
  worker.start()

for experiment in experiment_collection:
  enclosure_queue.put((experiment, otherVar))

enclosure_queue.join()

队列函数是这样的:

def run_experiment(i, q):
  while True:
    ... do stuff ...
    q.task_done()

---

我的新代码是这样的:

from multiprocessing import Process, Queue

num_fetch_threads = 4
enclosure_queue = Queue()

for i in range(num_fetch_threads):
  worker = Process(target=run_experiment, args=(i, enclosure_queue))
  worker.daemon = True
  worker.start()

for experiment in experiment_collection:
  enclosure_queue.put((experiment, otherVar))

worker.join() ## I only put this here bc enclosure_queue.join() is not available

以及新的队列函数:

def run_experiment(i, q):
  while True:
    ... do stuff ...
    ## not sure what should go here

---

我一直在阅读文档和谷歌，但无法弄清楚我错过了什么 - 我知道 task_done/join不属于 multiprocessing Queue类，但不清楚我应该使用什么。

"They differ in that Queue lacks the task_done() and join() methods introduced into Python 2.5’s Queue.Queue class." Source

但如果没有其中任何一个，我不确定队列如何知道它已完成，以及如何继续执行该程序。

Answer 1

考虑使用 multiprocessing.Pool而不是手动管理 worker 。 Pool 处理对 worker 的调度任务，具有 map 和 apply 等方便的功能，并支持 .close 和 .join 方法。 Pool 负责处理进程之间的队列并处理结果。以下是您的代码使用 multiprocessing.Pool 时的样子:

from multiprocessing import Pool

def do_experiment(exp):
    # run the experiment `exp`, will be called by `p.map`
    return result

p = Pool() # automatically scales to the number of CPUs available

results = p.map(do_experiment, experiment_collection)
p.close()
p.join()