如何使用Java1.5 BlockingQueue为多个使用者创建停止条件？

Question

对于单个生产者-消费者安排，生产者将“完成信号”作为队列中的最后一项。使用者检查从队列中取出的每个对象，并在done信号出队时关闭。

while(true) {
    MyQueueableObject o = queue.take() ;
    if ( o.type==PRODUCER_DONE_SIGNAL ) {
        return() ;
    }
    process ( o ) ;
}

然而，这只适用于单个生产者-消费者。因为BlockingQueue会阻塞take()调用，直到队列上有东西，所以在我看来，在多使用者设计中，应该有另一个嵌套循环来检查某些停止条件。

while(true) {
    while ( ... stop condition not detected ... ) {
        process ( queue.take()  ) ;
    }
    return ;
}

我已经详细考虑过这一点，我能想到的所有停止条件都涉及生产者线程的状态和队列的状态。两者看起来都不必要地复杂且容易出错。更糟糕的是，这种方法看起来像是一颗定时炸弹。在检查停止条件并获取false之后，在尝试从队列中获取某些内容之前，使用者线程可能会中断。在该中断期间，其他消费者可以清空队列。在这种情况下，队列将永远阻塞线程。

如何使用Java1.5 BlockingQueue为多个使用者创建停止条件？

Answer 1

在读取PRODUCER_DONE_SIGNAL之后，每个消费者都可以将其重新添加到队列中，以供下一个消费者使用。然而，我也会考虑使用队列外部的条件(例如，条件对象)：这样，即使队列仍然包含空的，您也可以提前终止线程(例如，如果您希望立即关闭)。

Answer 2

如果多次输入PRODUCER_DONE_SIGNAL，等于生产者的数量，该如何处理呢？