Как вы создаете условие остановки для нескольких потребителей, используя Java 1.5 BlockingQueue?

Question

Для одной схемы «производитель-потребитель» производитель помещает «сигнал готовности» в качестве последнего элемента в очередь. Потребитель проверяет каждый объект, взятый из очереди, и выключается при снятии сигнала готовности.

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 ;
}

Я долго об этом думал, и все условия остановки, которые я могу придумать, включают состояние Потоки производителя и состояние очереди. И то, и другое кажется излишне сложным и подверженным ошибкам. Еще хуже, этот подход кажется бомбой замедленного действия. Поток потребителя может быть прерван после проверки условия остановки и ложь и прежде чем пытаться взять что-то из очереди. Во время этого перерыва другие потребители мог бы опустошить очередь. В этом случае очередь навсегда заблокирует поток.

Как создать условие остановки для нескольких потребителей с помощью Java 1.5 BlockingQueue?

Answer 1

После прочтения PRODUCER_DONE_SIGNAL каждый потребитель может снова добавить его в очередь для следующего потребителя. Однако я также хотел бы рассмотреть возможность использования условия, внешнего по отношению к очереди (например, объекта условия): таким образом, вы можете завершить потоки досрочно, даже если очередь все еще содержит пустые данные (например, если вы хотите немедленно отключиться).

Answer 2

Как насчет подачи PRODUCER_DONE_SIGNAL несколько раз, равного количеству производителей?