BoundedPriorityBlockingQueue - thread safe, blocking and bounded?

Question

is there a priority-queue in Java which acts like LinkedBlockingQueue?

PriorityBlockingQueue is not blocking because it is unbounded.

Answer 1

If you don't need a full-blown BlockingQueue interface implementation then you can use Semaphore and something like this (in Kotlin):

interface BlockingBag<E: Any> {
    @Throws(InterruptedException::class)
    fun put(element: E)
    @Throws(InterruptedException::class)
    fun take(): E
}
class BlockingPriorityBag<E: Any>(val capacity: Int) : BlockingBag<E> {
    init {
        require(capacity >= 1) { "$capacity must be 1 or greater" }
    }
    private val queue = PriorityBlockingQueue<E>()
    private val semaphore = Semaphore(capacity)
    override fun take(): E {
        val item = queue.take()
        semaphore.release()
        return item
    }
    override fun put(element: E) {
        semaphore.acquire()
        queue.put(element)
    }
}

Answer 2

You can try MinMaxPriorityQueue from Google Guava and set a maximum size as next:

Queue<User> users = Queues.synchronizedQueue(
    MinMaxPriorityQueue.orderedBy(userComparator)
       .maximumSize(1000)
       .create()
);

NB: As a MinMaxPriorityQueue is not thread safe you need to use the decorator Queues.synchronizedQueue(Queue) allowing to make it thread safe.

As you need a BlockingQueue you will have to implement the decorator by yourself which is not hard to implement.

Here is how it should look like:

public class SynchronizedBlockingQueue implements BlockingQueue {

    private final BlockingQueue queue;

    public SynchronizedBlockingQueue(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public synchronized boolean add(final Object o) {
        return this.queue.add(o);
    }

    @Override
    public synchronized boolean offer(final Object o) {
        return this.offer(o);
    }
    ...
}

Then the code to create your BlockingQueue will be:

BlockingQueue<User> users = new SynchronizedBlockingQueue(
    MinMaxPriorityQueue.orderedBy(userComparator)
       .maximumSize(1000)
       .create()
);