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

Question

直接来自 Python docs :

class multiprocessing.Queue([maxsize])

...

qsize() Return the approximate size of the queue. Because of multithreading/multiprocessing semantics, this number is not reliable.

empty() Return True if the queue is empty, False otherwise. Because of multithreading/multiprocessing semantics, this is not reliable.

根据经验，我发现对于 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 是这样写的:

empty() Return True if the queue is empty, False otherwise.

完全没有提到可靠性。 SimpleQueue.empty() 可靠吗？

其次是 multiprocessing.JoinableQueue 可靠还是因为 task_done() 机制比 Queue 可靠？

这样的方法是否可以被认为是正确的，或者带有回调的方法(通过子节点之间的共享管道端点)是否更合适？

Answer 1

这不是一个直接的答案，但我已经开始越来越依赖于在保护条件下迭代输入队列。 multiprocessing 模块的文档中有一个示例:

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

因此，当您对队列的输入完成时，您只需将尽可能多的STOP字符串，或您选择的任何守卫，放入队列中，就像您启动进程一样。