Bounded PriorityBlockingQueue

Question

PriorityBlockingQueue is unbounded, but I need to bound it somehow. What is the best way to achieve that?

For information, the bounded PriorityBlockingQueue will be used in a ThreadPoolExecutor.

NB: By bounded I don't want to throw Exception if that happens, I want to put the object in the queue and then cut it based on its priority value. Is there any good way to do this cut thingie?

Answer 1

Of the top of my head, I'd subclass it and overwrite the put method to enforce this. If it goes over throw an exception or do whatever seems appropriate.

Something like:

public class LimitedPBQ extends PriorityBlockingQueue {

    private int maxItems;
    public LimitedPBQ(int maxItems){
        this.maxItems = maxItems;
    }

    @Override
    public boolean offer(Object e) {
        boolean success = super.offer(e);
        if(!success){
            return false;
        } else if (this.size()>maxItems){
            // Need to drop last item in queue
            // The array is not guaranteed to be in order, 
            // so you should sort it to be sure, even though Sun's Java 6 
            // version will return it in order
            this.remove(this.toArray()[this.size()-1]);
        }
        return true;
    }
}

Edit: Both add and put invoke offer, so overriding it should be enough

Edit 2: Should now remove the last element if over maxItems. There may be a more elegant way of doing it though.

Answer 2

I'd do something very similar to Kris, but maybe I'd insert first and then evict the lowest-priority item from the queue (add some QoS to it)

Answer 3

I actually wouldn't subclass it. While I can't put together example code right now, I'd suggest a version of the decorator pattern.

Create a new class and implement the interfaces implemented by your class of interest: PriorityBlockingQueue. I've found the following interfaces used by this class:

Serializable, Iterable<E>, Collection<E>, BlockingQueue<E>, Queue<E>

In the constructor for a class, accept a PriorityBlockingQueue as a constructor parameter.

Then implement all the methods required by the interfaces via the instances of the PriorityblockingQueue. Add any code required to make it Bounded.

Here's a quick diagram I put together in Violet UML:

Answer 4

If the order of the Runnables you want to execute is not strict (as is: it may occur that some lower priority tasks are executed even though higher priority tasks exist), then I would suggest the following, which boils down to periodically cutting the PriorityQueue down in size:

if (queue.size() > MIN_RETAIN * 2){
    ArrayList<T> toRetain = new ArrayList<T>(MIN_RETAIN);
    queue.drainTo(toRetain, MIN_RETAIN);
    queue.clear();
    for (T t : toRetain){
      queue.offer(t);
    }
}

This will obviously fail if the order needs to be strict, as draining will lead to a moment, wenn low priority task will retrieved from the queue using concurrent access.

The advantages are, that this is thread-safe and likely to get as fast as you can do with the priority queue design.

Answer 5

Have a look at the ForwardingQueue from the Google Collections API. For blocking semantics you could use a Semaphore.