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

【问题标题】：In python multi-producer & multi-consumer threading, may queue.join() be unreliable?在 python 多生产者和多消费者线程中，queue.join() 可能不可靠吗？
【发布时间】：2023-03-11 19:40:01
【问题描述】：

一个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() 这样不可靠，我如何检查所有工作是否完成？

【问题讨论】：

thread.join 和 Queue.join 测试不同的条件，应该用于不同的目的。 thread.join 只有在线程清空队列后退出时才有用。
Queue.join() 对其预期用途是可靠的：当所有工作项在消费者开始工作之前排队时。然后（并且只有那时）可以确定“所有标记为完成的队列项目”意味着“所有工作都已完成”。
感谢@TimPeters，但如果生产者的太多项目可能大于 RAM，则不可能在消费者开始之前排队 ，如何处理多生产者和多消费者线程？
@vbem，查看我的答案及其最近的编辑

标签： python multithreading queue

【解决方案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() 将在队列达到该大小时阻塞，直到消费者从队列中删除某些内容。

【讨论】：

不错！队列中有哨兵，似乎threading.Lock，threading.RLock等，不再需要了。
如果直接使用 Queue 就足够了，请务必使用它。那时，线程将是简单、明显和健壮的。其他噱头更适合其他任务。例如，虽然Queue 可以（以一种扭曲的方式）用于强制互斥（例如，确保只有一个线程正在写入控制台），但使用with some_Lock: 块。
确保在你的消费者退出之前调用 queue_of_data.task_done() ，否则 queue_of_data.join() 将永远挂起。 if data is None: queue_of_data.task_done() break
@hackerfriendly，我建议的模式根本不使用queue.join() - 所以也不需要使用queue.task_done()。