"Closing" a blocking queue

Question

I’m using java.util.concurrent.BlockingQueue in a very simple producer-consumer scenario. E.g. this pseudo code depicts the consumer part:

class QueueConsumer implements Runnable {      @Override     public void run() {         while(true)         {             try {                 ComplexObject complexObject = myBlockingQueue.take();                 //do something with the complex object             } catch (InterruptedException e) {                 Thread.currentThread().interrupt();             }         }     } } 

So far so good. In the javadoc of the blocking queue I read:

A BlockingQueue does not intrinsically support any kind of "close" or "shutdown" operation to indicate that no more items will be added. The needs and usage of such features tend to be implementation-dependent. For example, a common tactic is for producers to insert special end-of-stream or poison objects, that are interpreted accordingly when taken by consumers.

Unfortunately because of the generics in use and the nature of ComplexObject it’s not trivial to push a "poison object" into the queue. So this "common tactic" is not really convenient in my scenario.

My question is: what other good tactics/patterns can I use to "close" the queue?

Thank you!

Answer 1

If you have a handle to the consumer thread, you can interrupt it. With the code you gave, that will kill the consumer. I would not expect the producer to have this; it would probably have to callback to the program controller somehow to let it know it's done. Then the controller would interrupt the consumer thread.

You can always finish doing work before obeying the interrupt. For instance:

class QueueConsumer implements Runnable {     @Override     public void run() {         while(!(Thread.currentThread().isInterrupted())) {             try {                 final ComplexObject complexObject = myBlockingQueue.take();                 this.process(complexObject);              } catch (InterruptedException e) {                 // Set interrupted flag.                 Thread.currentThread().interrupt();             }         }          // Thread is getting ready to die, but first,         // drain remaining elements on the queue and process them.         final LinkedList<ComplexObject> remainingObjects;         myBlockingQueue.drainTo(remainingObjects);         for(ComplexObject complexObject : remainingObjects) {             this.process(complexObject);         }     }      private void process(final ComplexObject complexObject) {         // Do something with the complex object.     } } 

I would actually prefer that to somehow poisoning the queue anyway. If you want to kill the thread, ask the thread to kill itself.

(It's nice to see someone handling InterruptedException properly.)

---

There seems to be some contention about the handling of interruptions here. First, I would like everyone to read this article.

Now, with the understanding that no one actually read that, here's the deal. A thread will only receive an InterruptedException if it was currently blocking at the time of interrupt. In this case, Thread.interrupted() will return false. If it was not blocking, it will NOT receive this exception, and instead Thread.interrupted() will return true. Therefore, your loop guard should absolutely, no matter what, check Thread.interrupted(), or otherwise risk missing an interruption to the thread.

So, since you are checking Thread.interrupted() no matter what, and you are forced to catch InterruptedException (and should be dealing with it even if you weren't forced to), you now have two code areas which handle the same event, thread interruption. One way to handle this is normalize them into one condition, meaning either the boolean state check can throw the exception, or the exception can set the boolean state. I choose the later.

---

Edit: Note that the static Thread#interrupted method clears the the interrupted status of the current thread.

Answer 2

Another idea for making this simple:

class ComplexObject implements QueueableComplexObject {     /* the meat of your complex object is here as before, just need to      * add the following line and the "implements" clause above      */     @Override public ComplexObject asComplexObject() { return this; } }  enum NullComplexObject implements QueueableComplexObject {     INSTANCE;      @Override public ComplexObject asComplexObject() { return null; } }  interface QueueableComplexObject {     public ComplexObject asComplexObject(); } 

Then use BlockingQueue<QueueableComplexObject> as the queue. When you wish to end the queue's processing, do queue.offer(NullComplexObject.INSTANCE). On the consumer side, do

boolean ok = true; while (ok) {     ComplexObject obj = queue.take().asComplexObject();     if (obj == null)         ok = false;     else         process(obj); }  /* interrupt handling elided: implement this as you see fit,  * depending on whether you watch to swallow interrupts or propagate them  * as in your original post  */ 

No instanceof required, and you don't have to construct a fake ComplexObject which may be expensive/difficult depending on its implementation.