Python 多处理队列可靠性，Queue vs SimpleQueue vs JoinableQueue

Question

直接来自 Python docs:

类 multiprocessing.Queue([maxsize])

...

qsize() 返回队列的大致大小。由于多线程/多处理语义，这个数字是不可靠的。

空() 如果队列为空，则返回 True，否则返回 False。由于多线程/多处理语义，这是不可靠的。

我凭经验发现这对于Queue 来说是完全正确的，尤其是对于empty()。

在我的代码中，我有一堆进程（每个进程都是同一个主进程的子进程），每个进程的 run 方法中都有以下内容：

while self.active:
    if(self.exclusive_queue.empty() and self.exclusive_queue.qsize() == 0):
        try:
            self.exclusive_queue.put(self.general_queue.get(timeout=self.queue_timeout))
        except Queue.Empty as empty_queue:
            continue
    else:
        task = self.exclusive_queue.get()
        self.compute(task)

基本上，该进程等待general_queue 工作，但首先检查它的exclusive_queue。主进程可以将任务放入进程的通用队列或独占队列中。现在，在if(self.exclusive_queue.empty() and self.exclusive_queue.qsize() == 0) 中，我首先使用了self.exclusive_queue.empty()，这导致了相当奇怪的行为（qsize() 30+ 和empty() = True）。

所以我要说的是 - 文档中的 multiprocessing.queues.SimpleQueue 是这样写的：

空() 如果队列为空，则返回 True，否则返回 False。

根本没有提到可靠性。 SimpleQueue.empty() 可靠吗？

第二个是multiprocessing.JoinableQueue 可靠还是比Queue“更”可靠，因为task_done() 机制？

这种方法可以被认为是正确的，还是带有回调的方法（通过子级之间的共享管道端点）更合适？

Answer 1

不是一个直接的答案，但我已经开始越来越依赖于使用保护条件迭代输入队列。多处理模块的文档中有一个例子：

def worker(input, output):
    for func, args in iter(input.get, 'STOP'):
        result = calculate(func, args)
        output.put(result)

因此，当您对队列的输入完成后，您只需将 put 与您启动进程时一样多的 STOP 字符串或您选择的任何保护器添加到队列中。