BlockingQueue - заблокированные методы dumpTo ()

Question

В BlockingQueue есть метод, который называется СтокТо (), но он не заблокирован. Мне нужна очередь, которую я хочу заблокировать, но также могу получать объекты из очереди одним способом.

Object first = blockingQueue.take();

if ( blockingQueue.size() > 0 )
    blockingQueue.drainTo( list );

Полагаю, приведенный выше код будет работать, но я ищу элегантное решение.

Answer 1

Вы ссылаетесь на комментарий в JavaDoc :

Далее, поведение этой операции не определено, если указанная коллекция изменяется во время выполнения операции.

Я считаю, что это относится к коллекции list в вашем примере:

blockingQueue.drainTo(list);

означает, что вы не можете изменить list в то же время, когда вы сливаете из blockingQueue в list. Однако очередь блокировки внутренне синхронизируется, поэтому, когда вызывается drainTo, ставится и блокируется (см. Примечание ниже). Если бы он этого не делал, то он не был бы действительно поточно-ориентированным. Вы можете взглянуть на исходный код и убедиться, что drainTo является поточно-ориентированным в отношении самой очереди блокировки.

Альтернативно, вы имеете в виду, что когда вы вызываете drainTo, вы хотите, чтобы он блокировался, пока хотя бы один объект не был добавлен в очередь? В этом случае у вас мало выбора, кроме:

list.add(blockingQueue.take());
blockingQueue.drainTo(list);

блокировать до тех пор, пока не будет добавлен один или несколько элементов, а затем слить всю очередь в коллекцию list.

Примечание. Начиная с Java 7, для получения и размещения используется отдельная блокировка. Операции «пут» теперь разрешены во время «слива» (и ряда других операций «бери»).

Answer 2

Если вы используете Google Guava, есть отличный метод Queues.drain().

Сливает очередь как BlockingQueue.drainTo(Collection, int), но если Запрошенные numElements элементы недоступны, будут ждать до указанного времени ожидания.

Answer 3

Я нашел этот шаблон полезным.

List<byte[]> blobs = new ArrayList<byte[]>();
if (queue.drainTo(blobs, batch) == 0) {
   blobs.add(queue.take());
}

Answer 4

Исходный код:

 596:     public int drainTo(Collection<? super E> c) {
              //arg. check
 603:         lock.lock();
 604:         try {
 608:             for (n = 0 ; n != count ; n++) {
 609:                 c.add(items[n]);
 613:             }
 614:             if (n > 0) {
 618:                 notFull.signalAll();
 619:             }
 620:             return n;
 621:         } finally {
 622:             lock.unlock();
 623:         }
 624:     }

ArrayBlockingQueue стремится вернуть 0. Кстати, он может сделать это, прежде чем взять блокировку

Answer 5

С доступным API я не думаю, что вы станете намного элегантнее. Кроме вы можете удалить размер теста.

Если вы хотите атомарно получить непрерывную последовательность элементов, даже если совпадает другая операция удаления, я не верю, что даже drainTo гарантирует это.