使用多进程避免因队列溢出而导致的死锁。JoinableQueue

问题描述

假设我们有一个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()阻塞。

有没有办法避免这样的拥堵？

---

解决方案

根据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。