"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.

Answer 3

An alternative would be to wrap the processing you're doing with an ExecutorService, and let the ExecutorService itself control whether or not jobs get added to the queue.

Basically, you take advantage of ExecutorService.shutdown(), which when called disallows any more tasks from being processed by the executor.

I'm not sure how you're currently submitting tasks to the QueueConsumer in your example. I've made the assumption that you have some sort of submit() method, and used a similar method in the example.

import java.util.concurrent.*;

class QueueConsumer {
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public void shutdown() {
        executor.shutdown(); // gracefully shuts down the executor
    }

    // 'Result' is a class you'll have to write yourself, if you want.
    // If you don't need to provide a result, you can just Runnable
    // instead of Callable.
    public Future<Result> submit(final ComplexObject complexObject) {
        if(executor.isShutdown()) {
            // handle submitted tasks after the executor has been told to shutdown
        }

        return executor.submit(new Callable<Result>() {
            @Override
            public Result call() {
                return process(complexObject);
            }
        });
    }

    private Result process(final ComplexObject complexObject) {
        // Do something with the complex object.
    }
}

This example is just an off-the-cuff illustration of what the java.util.concurrent package offers; there are probably some optimizations that could be made to it (e.g., QueueConsumer as its own class probably isn't even necessary; you could just provide the ExecutorService to whatever producers are submitting the tasks).

Dig through the java.util.concurrent package (starting at some of the links above). You might find that it gives you a lot of great options for what you're trying to do, and you don't even have to worry about regulating the work queue.

Answer 4

Another possibility for making a poison object: Make it be a particular instance of the class. This way, you do not have to muck around subtypes or screw up your generic.

Drawback: This won't work if there's some sort of serialization barrier between the producer and consumer.

public class ComplexObject
{
    public static final POISON_INSTANCE = new ComplexObject();

    public ComplexObject(whatever arguments) {
    }

    // Empty constructor for creating poison instance.
    private ComplexObject() {
    }
}

class QueueConsumer implements Runnable {
    @Override
    public void run() {
        while(!(Thread.currentThread().interrupted())) {
            try {
                final ComplexObject complexObject = myBlockingQueue.take();
                if (complexObject == ComplexObject.POISON_INSTANCE)
                    return;

                // Process complex object.

            } catch (InterruptedException e) {
                // Set interrupted flag.
                Thread.currentThread().interrupt();
            }
        }
    }
}

Answer 5

You can wrap your generic object into a dataobject. On this dataobject you can add additional data like the poison object status. The dataobject is a class with 2 fields. T complexObject; and boolean poison;.

Your consumer takes the data objects from the queue. If a poison object is returned, you close the consumer, else you unwrap the generic and call 'process(complexObject)'.

I'm using a java.util.concurrent.LinkedBlockingDeque<E> so that you can add object at the end of the queue and take them from the front. That way your object will be handled in order, but more important it's safe to close the queue after you run into the poison object.

To support multiple consumers, I add the poison object back onto the queue when I run into it.

public final class Data<T> {
    private boolean poison = false;
    private T complexObject;

    public Data() {
        this.poison = true;
    }

    public Data(T complexObject) {
        this.complexObject = complexObject;
    }

    public boolean isPoison() {
        return poison;
    }

    public T getComplexObject() {
        return complexObject;
    }
}

public class Consumer <T> implements Runnable {

    @Override
    public final void run() {
        Data<T> data;
        try {
            while (!(data = queue.takeFirst()).isPoison()) {
                process(data.getComplexObject());
            }
        } catch (final InterruptedException e) {
            Thread.currentThread().interrupt();
            return;
        }
        // add the poison object back so other consumers can stop too.
        queue.addLast(line);
    }
}

Answer 6

Is it possible to extend ComplexObject and mock out the non-trivial creation functionality? Essentially you're ending up with a shell object but you can do then do instance of to see if is the end of queue object.

Answer 7

It seems reasonable to me to implement a close-able BlockingQueue:

import java.util.concurrent.BlockingQueue;

public interface CloseableBlockingQueue<E> extends BlockingQueue<E> {
    /** Returns <tt>true</tt> if this queue is closed, <tt>false</tt> otherwise. */
    public boolean isClosed();

    /** Closes this queue; elements cannot be added to a closed queue. **/
    public void close();
}

It would be quite straight forward to implement this with the following behaviours (cf. the methods summary table):

• Insert:

  • Throws exception, Special value:

    Behaves like a full Queue, caller's responsibility to test isClosed().

  • Blocks:

    Throws IllegalStateException if and when closed.

  • Times out:

    Returns false if and when closed, caller's responsibility to test isClosed().

• Remove:

  • Throws exception, Special value:

    Behaves like a empty Queue, caller's responsibility to test isClosed().

  • Blocks:

    Throws NoSuchElementException if and when closed.

  • Times out:

    Returns null if and when closed, caller's responsibility to test isClosed().

• Examine

  No change.

I did this by editing the source, find it at github.com.

Answer 8

Today I solved this problem using a wrapper object. Since the ComplexObject is too complex to subclass I wrapped the ComplexObject into ComplexObjectWrapper object. Then used ComplexObjectWrapper as the generic type.

public class ComplexObjectWrapper {
ComplexObject obj;
}

public class EndOfQueue extends ComplexObjectWrapper{}

Now instead of BlockingQueue<ComplexObject> I did BlockingQueue<ComplexObjectWrapper>

Since I had control of both the Consumer and Producer this solution worked for me.

Answer 9

I have used this system:

ConsumerClass
private boolean queueIsNotEmpty = true;//with setter
...
do {
    ...
    sharedQueue.drainTo(docs);
    ...
} while (queueIsNotEmpty || sharedQueue.isEmpty());

When producer finish, I set on consumerObject, queueIsNotEmpty field to false

Answer 10

In this situation, you generally have to ditch the generics and make the queue hold type Object. then, you just need check for your "poison" Object before casting to the actual type.