BlockingQueue - blocked drainTo() methods

Question

BlockingQueue has the method called drainTo() but it is not blocked. I need a queue that I want to block but also able to retrieve queued objects in a single method.

Object first = blockingQueue.take();

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

I guess the above code will work but I'm looking for an elegant solution.

Answer 1

Are you referring to the comment in the JavaDoc:

Further, the behavior of this operation is undefined if the specified collection is modified while the operation is in progress.

I believe that this refers to the collection list in your example:

blockingQueue.drainTo(list);

meaning that you cannot modify list at the same time you are draining from blockingQueue into list. However, the blocking queue internally synchronizes so that when drainTo is called, puts and (see note below) gets will block. If it did not do this, then it would not be truly Thread-safe. You can look at the source code and verify that drainTo is Thread-safe regarding the blocking queue itself.

Alternately, do you mean that when you call drainTo that you want it to block until at least one object has been added to the queue? In that case, you have little choice other than:

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

to block until one or more items have been added, and then drain the entire queue into the collection list.

Note: As of Java 7, a separate lock is used for gets and puts. Put operations are now permitted during a drainTo (and a number of other take operations).

Answer 2

If you happen to use Google Guava, there's a nifty Queues.drain() method.

Drains the queue as BlockingQueue.drainTo(Collection, int), but if the requested numElements elements are not available, it will wait for them up to the specified timeout.

Answer 3

I found this pattern useful.

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