python - 在python多生产者和多消费者线程中，queue.join()可能不可靠吗？

Question

Python 多生产者和多消费者线程伪代码:

def threadProducer():
    while upstreams_not_done:
        data = do_some_work()
        queue_of_data.put(data)

def threadConsumer():
    while True:
        data = queue_of_data.get()
        do_other_work()
        queue_of_data.task_done()

queue_of_data = queue.Queue()

list_of_producers = create_and_start_producers()
list_of_consumers = create_and_start_consumers()

queue_of_data.join()
# is now all work done?

其中为队列中的每个项目调用 queue_of_data.task_done()。

当生产者工作速度比消费者慢时，是否有可能queue_of_data.join() 在没有生产者生成数据的某个时刻非阻塞，但是所有消费者都通过task_done()完成他们的任务？

如果 Queue.join() 像这样不可靠，我如何检查所有工作是否完成？

Answer 1

通常的方法是在生产者完成时将一个哨兵值(例如None)放入队列中，每个消费者线程一个。然后，当线程从队列中提取 None 时，消费者将被写入退出线程。

因此，例如在主程序中:

for t in list_of_producers:
    t.join()
# Now we know all producers are done.
for t in list_of_consumers:
    queue_of_data.put(None)  # tell a consumer we're done
for t in list_of_consumers:
    t.join()

消费者看起来像:

def threadConsumer():
    while True:
        data = queue_of_data.get()
        if data is None:
            break
        do_other_work()

注意:如果生产者可以压垮消费者，请创建具有最大大小的队列。然后，当队列达到该大小时，queue.put() 将阻塞，直到消费者从队列中删除某些内容。