PriorityBlockingQueue: Thread-Safe Priority Queue (+ Example)

In this article, you will learn how PriorityBlockingQueue works and what characteristics it has. Examples will show you how to use it – and how, as the work queue of a ThreadPoolExecutor, it runs tasks by priority.

Here we are in the class hierarchy:

PriorityBlockingQueue Characteristics

The java.util.concurrent.PriorityBlockingQueue is a thread-safe and blocking variant of the PriorityQueue. In the linked article, you will also learn what a priority queue is.

As with PriorityQueue, the elements are stored in an array representing a min-heap. So the smallest element is always at the head of the queue.

The iterator, however, does not return the elements in priority order but in the order of the internal heap array – the same goes for toString() and toArray(). To get them sorted, take them out with take(), poll(), or drainTo().

A single ReentrantLock ensures thread safety (see below).

PriorityBlockingQueue is not bounded, so it has no capacity limit. That means that put(e) and offer(e, time, unit) never block. Only the dequeue operations take() and poll(time, unit) block when the queue is empty.

Its only limit is memory: if there is not enough left to grow the array, an insertion can fail with an OutOfMemoryError, according to the Javadoc.

null elements are not allowed (put(null) throws a NullPointerException).

The characteristics in detail:

Underlying data structure	Thread-safe?	Blocking/ non-blocking	Fairness policy	Bounded/ unbounded	Iterator type
Min-heap (stored in an array)	Yes (pessimistic locking with a lock)	Blocking (only dequeue)	Not available	Unbounded	Weakly consistent¹

¹ Weakly consistent: According to the Javadoc, the iterator may reflect changes made after its creation, but need not. In fact, it reflects none: it works on a copy of the array that iterator() creates when called, returns every element that is in the queue at that moment exactly once, and never throws a ConcurrentModificationException.

The iterator supports remove(). The method removes the element from the queue itself, not just from the copy.

Recommended Use Case

I recommend PriorityBlockingQueue when several threads exchange elements by priority rather than in insertion order – for example, tasks that a thread pool should run by urgency. The section “PriorityBlockingQueue in a ThreadPoolExecutor” shows how to do that with a ThreadPoolExecutor.

All operations go through the same ReentrantLock, though – under high contention (many threads accessing the queue at the same time), producers and consumers wait for each other.

And because the queue is unbounded, it never slows down a producer. If elements go in faster than they are taken out, the queue grows until memory runs out.

If you don’t need priorities but insertion order, the LinkedBlockingQueue with its two locks is usually the better choice; if you need a capacity limit, the ArrayBlockingQueue.

PriorityBlockingQueue Example

The following example shows how to create a PriorityBlockingQueue and how multiple threads read and write to it (→ code on GitHub).

Reading threads run every 3 seconds, starting immediately after the queue is created.

Writing threads start after 3.5 seconds (so that two reading threads are already waiting) and write a random value to the queue every second.

public class PriorityBlockingQueueExample {
  private static final long startTime = System.currentTimeMillis();

  static void main() throws InterruptedException {
    BlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);

    // Start reading from the queue immediately, every 3 seconds
    for (int i = 0; i < 8; i++) {
      int delaySeconds = i * 3;
      pool.schedule(() -> dequeue(queue), delaySeconds, TimeUnit.SECONDS);
    }

    // Start writing to the queue after 3.5 seconds (so there are already 2 threads
    // waiting), every second (so that the queue fills faster than it's emptied,
    // so that we see some more elements and their order in the queue)
    for (int i = 0; i < 8; i++) {
      int delayMillis = 3500 + i * 1000;
      pool.schedule(() -> enqueue(queue), delayMillis, TimeUnit.MILLISECONDS);
    }

    pool.shutdown();
    pool.awaitTermination(1, TimeUnit.MINUTES);
  }

  private static void enqueue(BlockingQueue<Integer> queue) {
    int element = ThreadLocalRandom.current().nextInt(10, 100);
    log("Calling queue.put(%d) (queue = %s)...", element, queue);
    try {
      queue.put(element);
      log("queue.put(%d) returned (queue = %s)", element, queue);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void dequeue(BlockingQueue<Integer> queue) {
    log("    Calling queue.take() (queue = %s)...", queue);
    try {
      Integer element = queue.take();
      log("    queue.take() returned %d (queue = %s)", element, queue);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void log(String format, Object... args) {
    System.out.printf(
        Locale.US,
        "[%4.1fs] [%-16s] %s%n",
        (System.currentTimeMillis() - startTime) / 1000.0,
        Thread.currentThread().getName(),
        String.format(format, args));
  }
}

Below you can see an example output of the program:

[ 0.0s] [pool-1-thread-1 ]     Calling queue.take() (queue = [])...
[ 3.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [])...
[ 3.5s] [pool-1-thread-6 ] Calling queue.put(87) (queue = [])...
[ 3.5s] [pool-1-thread-6 ] queue.put(87) returned (queue = [])
[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 87 (queue = [])
[ 4.5s] [pool-1-thread-9 ] Calling queue.put(89) (queue = [])...
[ 4.5s] [pool-1-thread-9 ] queue.put(89) returned (queue = [])
[ 4.5s] [pool-1-thread-2 ]     queue.take() returned 89 (queue = [])
[ 5.5s] [pool-1-thread-7 ] Calling queue.put(31) (queue = [])...
[ 5.5s] [pool-1-thread-7 ] queue.put(31) returned (queue = [31])
[ 6.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [31])...
[ 6.0s] [pool-1-thread-4 ]     queue.take() returned 31 (queue = [])
[ 6.5s] [pool-1-thread-5 ] Calling queue.put(71) (queue = [])...
[ 6.5s] [pool-1-thread-5 ] queue.put(71) returned (queue = [71])
[ 7.5s] [pool-1-thread-8 ] Calling queue.put(15) (queue = [71])...
[ 7.5s] [pool-1-thread-8 ] queue.put(15) returned (queue = [15, 71])
[ 8.5s] [pool-1-thread-10] Calling queue.put(33) (queue = [15, 71])...
[ 8.5s] [pool-1-thread-10] queue.put(33) returned (queue = [15, 71, 33])
[ 9.0s] [pool-1-thread-3 ]     Calling queue.take() (queue = [15, 71, 33])...
[ 9.0s] [pool-1-thread-3 ]     queue.take() returned 15 (queue = [33, 71])
[ 9.5s] [pool-1-thread-6 ] Calling queue.put(58) (queue = [33, 71])...
[ 9.5s] [pool-1-thread-6 ] queue.put(58) returned (queue = [33, 71, 58])
[10.5s] [pool-1-thread-1 ] Calling queue.put(19) (queue = [33, 71, 58])...
[10.5s] [pool-1-thread-1 ] queue.put(19) returned (queue = [19, 33, 58, 71])
[12.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [19, 33, 58, 71])...
[12.0s] [pool-1-thread-9 ]     queue.take() returned 19 (queue = [33, 71, 58])
[15.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [33, 71, 58])...
[15.0s] [pool-1-thread-2 ]     queue.take() returned 33 (queue = [58, 71])
[18.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [58, 71])...
[18.0s] [pool-1-thread-7 ]     queue.take() returned 58 (queue = [71])
[21.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [71])...
[21.0s] [pool-1-thread-4 ]     queue.take() returned 71 (queue = [])

What can we see in this sample output?

First of all, you see how after 0.0 s and 3.0 s, threads 1 and 2 block when calling take() because the queue is empty.

After 3.5 s, thread 6 writes the 87 into the queue. Immediately afterward, the previously blocked thread 1 wakes up again and takes the 87.

After 4.5 s, thread 9 writes the 89 into the queue, which is immediately taken out again by thread 2.

After 5.5 s, the 31 is written into the queue, which is taken out again after 6.0 s.

After 6.5 s, 7.5 s, and 8.5 s, the 71, the 15, and the 33 are written into the queue. You can see how the smallest element is always at the head (left) of the queue.

Behind the head, after 8.5 s, the 71 comes before the 33. That is because toString() prints the heap array – and the heap array is sorted only at its head. The diagram at the end of this section shows how tree and array relate.

After 9.0 s, the smallest element, the 15, is removed. The next smallest element, 33, is then placed at the head of the queue.

After 9.5 s and 10.5 s, two more elements, 58 and 19, are written to the queue. Again, you can see how the smallest element is at the queue’s head.

The queue now contains four elements. No other elements are written to the queue, and the existing elements are taken according to their priority.

After 12.0 s, thread 9 takes the 19, and the queue prints [33, 71, 58] – here, too, the 71 comes before the smaller 58.

The following diagram shows the heap before and after this take(), each time as a tree and as an array. The array holds the tree level by level, from left to right.

When the 19 is taken, the queue moves the last element of the array, the 71, to the root. Because the 71 is larger than its two children, 33 and 58, it swaps places with the smaller one, the 33 – and the smallest element is at the root again:

The article “Priority Queue Implementation in Java Using a Min-Heap” shows this so-called sift down step by step.

PriorityBlockingQueue with a Comparator

By default, PriorityBlockingQueue orders its elements by their natural ordering, that is, by compareTo(). The elements must then implement Comparable. The queue rejects an element that doesn’t with a ClassCastException on insertion.

You define a different order with a Comparator. Of the four constructors of PriorityBlockingQueue, only one accepts a Comparator, and that constructor also requires an initial capacity.

The following program creates a queue with initial capacity 11 and Comparator.reverseOrder(), writes three numbers into it, and takes them out again (→ code on GitHub):

public class PriorityBlockingQueueComparatorExample {
  static void main() throws InterruptedException {
    BlockingQueue<Integer> queue =
        new PriorityBlockingQueue<>(11, Comparator.reverseOrder());
    queue.put(31);
    queue.put(87);
    queue.put(15);
    IO.println(queue.take() + " " + queue.take() + " " + queue.take());
  }
}

The program prints the following:

87 31 15

With Comparator.reverseOrder(), the largest element is at the head – the min-heap becomes a max-heap.

The other three constructors take no Comparator:

• The constructor without parameters creates the array with 11 slots.
• The constructor with an int creates it with the given initial capacity. It rejects an initial capacity below 1 – like the constructor with a Comparator – with an IllegalArgumentException.
• The constructor with a collection takes over its elements. If the collection is a SortedSet or a PriorityBlockingQueue, it also takes over its ordering instead of the natural one.

Elements with Equal Priority

The queue does not define the order in which it returns elements of equal priority. Insertion order is not guaranteed either.

If you need that order, give each element a sequence number as a second sort key. The Javadoc of the class shows this with the class FIFOEntry: it draws the number from an AtomicLong when it is created and compares it when the elements themselves are equal.

PriorityBlockingQueue in a ThreadPoolExecutor

A ThreadPoolExecutor takes its work queue as a BlockingQueue in its constructor. With a PriorityBlockingQueue, it runs waiting tasks by priority instead of in the order in which they were submitted.

The following program submits four tasks of type PrioritizedTask to a pool with a single thread. The first task, E, runs at once. The other three wait in the queue, and the thread takes them out by priority – lowest number first (→ code on GitHub: PrioritizedTask, PriorityThreadPoolExample):

public record PrioritizedTask(int priority, String name)
    implements Runnable, Comparable<PrioritizedTask> {

  @Override
  public void run() {
    IO.println("Running task " + name + " (priority " + priority + ")");
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  @Override
  public int compareTo(PrioritizedTask other) {
    return Integer.compare(priority, other.priority);
  }
}

public class PriorityThreadPoolExample {
  static void main() throws InterruptedException {
    ThreadPoolExecutor pool = new ThreadPoolExecutor(
        1, 1, 0, TimeUnit.SECONDS, new PriorityBlockingQueue<>());

    pool.execute(new PrioritizedTask(5, "E")); // runs at once on the only thread
    pool.execute(new PrioritizedTask(3, "C")); // these three wait in the queue
    pool.execute(new PrioritizedTask(1, "A"));
    pool.execute(new PrioritizedTask(2, "B"));

    pool.shutdown();
    pool.awaitTermination(1, TimeUnit.MINUTES);
  }
}

In the listing, the constructor of the ThreadPoolExecutor receives five arguments:

• corePoolSize = 1: the number of threads the pool keeps permanently, even while they have nothing to do.
• maximumPoolSize = 1: the number of threads the pool may grow to.
• keepAliveTime = 0 and unit = TimeUnit.SECONDS: how long a thread beyond corePoolSize waits for new work before it terminates.
• workQueue: the PriorityBlockingQueue in which tasks wait while no thread is free.

The sleep() in run() keeps the thread busy with task E long enough for C, A, and B to arrive in the queue before it fetches the next task.

The program prints the following:

Running task E (priority 5)
Running task A (priority 1)
Running task B (priority 2)
Running task C (priority 3)

C was submitted before A and B and still runs last.

There is a reason why corePoolSize and maximumPoolSize are equal in the listing: the ThreadPoolExecutor starts additional threads beyond corePoolSize only when the queue rejects a task – and an unbounded queue like PriorityBlockingQueue never rejects one. A larger value for maximumPoolSize would therefore have no effect; the Javadoc of ThreadPoolExecutor says so, too.

I therefore recommend setting both values to the same number with a PriorityBlockingQueue.

ClassCastException with submit()

If you replace execute() with submit() in the listing, the program stops at the second call:

pool.submit(new PrioritizedTask(5, "E"));
pool.submit(new PrioritizedTask(3, "C"));

The second call throws the following exception:

Exception in thread "main" java.lang.ClassCastException: class java.util.concurrent.FutureTask cannot be cast to class java.lang.Comparable (java.util.concurrent.FutureTask and java.lang.Comparable are in module java.base of loader 'bootstrap')
    at java.base/java.util.concurrent.PriorityBlockingQueue.siftUpComparable(PriorityBlockingQueue.java:350)
    at java.base/java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:476)
    at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1320)
    at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:125)

The reason: submit() wraps the task in a FutureTask, which it returns as a Future, and passes it on to execute(). FutureTask does not implement Comparable, and the queue fails on the cast when sorting it in. The first call still succeeds because the pool hands the first task directly to its new thread, without using the queue.

If you don’t need a Future, pass the tasks to execute() as in the listing above. If you do need one – for example, to cancel a waiting task or to wait for its result –, override the method newTaskFor() in a subclass of ThreadPoolExecutor; it is the method that provides submit() with the FutureTask. The Javadoc of AbstractExecutorService explicitly provides for this.

Next to the record PrioritizedTask from above, the following listing contains three new classes (→ code on GitHub: PrioritizedFutureTask, PriorityThreadPoolExecutor, PriorityThreadPoolSubmitExample):

• PrioritizedFutureTask is a FutureTask that implements Comparable by comparing the priority of its task.
• PriorityThreadPoolExecutor returns such a PrioritizedFutureTask from newTaskFor().
• PriorityThreadPoolSubmitExample submits the tasks with submit() and cancels B through its Future while B is still waiting in the queue.

public record PrioritizedTask(int priority, String name)
    implements Runnable, Comparable<PrioritizedTask> {

  @Override
  public void run() {
    IO.println("Running task " + name + " (priority " + priority + ")");
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  @Override
  public int compareTo(PrioritizedTask other) {
    return Integer.compare(priority, other.priority);
  }
}

public class PrioritizedFutureTask<T> extends FutureTask<T>
    implements Comparable<PrioritizedFutureTask<?>> {
  private final PrioritizedTask task;

  PrioritizedFutureTask(PrioritizedTask task, T value) {
    super(task, value);
    this.task = task;
  }

  @Override
  public int compareTo(PrioritizedFutureTask<?> other) {
    return task.compareTo(other.task);
  }
}

public class PriorityThreadPoolExecutor extends ThreadPoolExecutor {
  PriorityThreadPoolExecutor(int poolSize) {
    super(poolSize, poolSize, 0, TimeUnit.SECONDS, new PriorityBlockingQueue<>());
  }

  @Override
  protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new PrioritizedFutureTask<>((PrioritizedTask) runnable, value);
  }
}

public class PriorityThreadPoolSubmitExample {
  static void main() throws InterruptedException {
    ThreadPoolExecutor pool = new PriorityThreadPoolExecutor(1);

    pool.submit(new PrioritizedTask(5, "E"));
    pool.submit(new PrioritizedTask(3, "C"));
    pool.submit(new PrioritizedTask(1, "A"));
    Future<?> taskB = pool.submit(new PrioritizedTask(2, "B"));

    taskB.cancel(false); // B is still waiting in the queue

    pool.shutdown();
    pool.awaitTermination(1, TimeUnit.MINUTES);
  }
}

The program prints only three lines – B is missing, and the other tasks still run by priority:

Running task E (priority 5)
Running task A (priority 1)
Running task C (priority 3)

The cancelled task stays in the queue until the thread takes it out. It is then no longer executed, as the Javadoc of ThreadPoolExecutor states. The source code of FutureTask.run() shows why: the method returns immediately because the task is no longer in its initial state.

The cast in newTaskFor() requires that the pool receives only PrioritizedTasks. And submit() with a Callable goes through the second variant of newTaskFor(), which takes a Callable instead of a Runnable; you would have to override it as well.

Is PriorityBlockingQueue Thread-Safe?

Yes, PriorityBlockingQueue is thread-safe.

Thread safety is provided by a single ReentrantLock, simply called lock in the source code. It guards all public operations, and one condition hangs off it: notEmpty, on which take() waits while the queue is empty. The class needs no second condition for a full queue – it never gets full.

Growing the array is the exception. When the array is full, offer() releases the lock and allocates the larger array under a simple spinlock: the int field allocationSpinLock, which is set from 0 to 1 by compare-and-set. Then offer() takes the lock again and copies the elements over.

This way, consumer threads can keep taking elements with take() while a producer thread allocates the new array. According to a comment in the source code, this prevents waiting consumers from being postponed again and again while elements pile up.

Below 64 slots, the array grows to a little more than double its size (11 slots become 24, then 50, then 102); from 64 slots on, it grows by half.

Summary and Outlook

In this article, you learned about PriorityBlockingQueue. It is thread-safe, blocking, and unbounded; a single ReentrantLock guards its min-heap. take() always returns the smallest element, while the iterator and toString() return the heap array in its internal order.

As the work queue of a ThreadPoolExecutor, it runs tasks by priority – directly with execute(), with submit() only once you override newTaskFor().

Starting with the next part of the tutorial series, I will introduce you to some queue implementations for special cases, beginning with the DelayQueue.