使用 multiprocessing.JoinableQueue 避免由于队列溢出导致的死锁

Question

假设我们有一个multiprocessing.Pool，其中工作线程共享一个multiprocessing.JoinableQueue，使工作项出队并可能使更多工作入队：

def worker_main(queue):
    while True:
        work = queue.get()
        for new_work in process(work):
            queue.put(new_work)
        queue.task_done()

当队列填满时，queue.put() 将阻塞。只要至少有一个进程从队列中读取queue.get()，它就会释放队列中的空间以解除对写入者的阻塞。但所有进程都可能同时在 queue.put() 处阻塞。

有没有办法避免像这样被卡住？

Answer 1

根据process(work) 创建更多项目的频率，除了无限最大大小的队列之外，可能没有其他解决方案。

简而言之，您的队列必须足够大，以容纳您随时可以拥有的整个积压工作项。

---

由于queue is implemented with semaphores，可能确实有a hard size limit of SEM_VALUE_MAX其中in MacOS is 32767。因此，如果这还不够，您将需要对该实现进行子类化或使用put(block=False) 并处理queue.Full（例如，将多余的项目放在其他地方）。

或者，查看one of the 3rd-party implementations of distributed work item queue for Python。