Java LinkedBlockingQueue: Blocking Queue With Two Locks

This part of the tutorial series is about LinkedBlockingQueue. You will get to know its unique characteristics and see how to use this queue with an example. You will also learn when exactly you should use this queue.

Here we are in the class hierarchy:

LinkedBlockingQueue Characteristics

The class java.util.concurrent.LinkedBlockingQueue is – just like ConcurrentLinkedQueue – based on a linked list, but is – like ArrayBlockingQueue presented in the next part – thread-safe (see below), blocking, and optionally bounded.

Unlike ArrayBlockingQueue, LinkedBlockingQueue does not provide a fairness policy. (Fairness policy means that blocking methods are served in the order they were called.)

Whether the queue is bounded is decided when you create it: new LinkedBlockingQueue<>() creates a queue with the capacity Integer.MAX_VALUE – practically unbounded. new LinkedBlockingQueue<>(capacity) sets a limit; once it is reached, put() blocks, and offer() returns false.

null elements are not allowed (offer(null) throws a NullPointerException) – which is why poll() can unambiguously signal “queue empty” with the return value null.

The queue’s characteristics in detail:

Underlying data structure	Thread-safe?	Blocking/ non-blocking	Fairness policy	Bounded/ unbounded	Iterator type
Linked list	Yes (pessimistic locking with two locks)	Blocking	Not available	Optionally bounded (default: unbounded)	Weakly consistent¹

¹ Weakly consistent: Elements that remain in the queue without interruption since the iterator was created are returned by the iterator exactly once. Changes after its creation – insertions as well as removals – may be reflected by the iterator, but need not be; it never throws a ConcurrentModificationException.

The iterator supports remove().

Recommended Use Case

I recommend LinkedBlockingQueue if you need a blocking, thread-safe queue.

Compared to ArrayBlockingQueue, it has the higher throughput in most cases because producers and consumers rarely block each other – how that works is shown in the section “Is LinkedBlockingQueue thread-safe?”. In return, it allocates a node of its own for every element, and according to the Javadoc, its performance is less predictable than that of an array-based queue in most concurrent applications.

ArrayBlockingQueue is the better choice if you need a fairness policy or if even latencies matter more to you than throughput. If nothing should block at all, use ConcurrentLinkedQueue.

LinkedBlockingQueue Example

The following example shows how to use LinkedBlockingQueue. We create a queue with a capacity of 3. Immediately afterward, we start reading elements from the queue at intervals of three seconds. After 3.5 seconds, we begin writing elements to the queue at intervals of one second each (→ code on GitHub).

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

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

    // Start reading from the queue immediately, every 3 seconds
    for (int i = 0; i < 10; 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 a full queue soon)
    for (int i = 0; i < 10; i++) {
      int element = i; // Assign to an effectively final variable
      int delayMillis = 3500 + i * 1000;
      pool.schedule(
          () -> enqueue(queue, element), delayMillis, TimeUnit.MILLISECONDS);
    }

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

  private static void enqueue(BlockingQueue<Integer> queue, int element) {
    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 the output of the sample program:

[ 0.0s] [pool-1-thread-1 ]     Calling queue.take() (queue = [])...
[ 3.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [])...
[ 3.5s] [pool-1-thread-8 ] Calling queue.put(0) (queue = [])...
[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 0 (queue = [])
[ 3.5s] [pool-1-thread-8 ] queue.put(0) returned (queue = [])
[ 4.5s] [pool-1-thread-5 ] Calling queue.put(1) (queue = [])...
[ 4.5s] [pool-1-thread-4 ]     queue.take() returned 1 (queue = [])
[ 4.5s] [pool-1-thread-5 ] queue.put(1) returned (queue = [])
[ 5.5s] [pool-1-thread-3 ] Calling queue.put(2) (queue = [])...
[ 5.5s] [pool-1-thread-3 ] queue.put(2) returned (queue = [2])
[ 6.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [2])...
[ 6.0s] [pool-1-thread-7 ]     queue.take() returned 2 (queue = [])
[ 6.5s] [pool-1-thread-9 ] Calling queue.put(3) (queue = [])...
[ 6.5s] [pool-1-thread-9 ] queue.put(3) returned (queue = [3])
[ 7.5s] [pool-1-thread-6 ] Calling queue.put(4) (queue = [3])...
[ 7.5s] [pool-1-thread-6 ] queue.put(4) returned (queue = [3, 4])
[ 8.5s] [pool-1-thread-2 ] Calling queue.put(5) (queue = [3, 4])...
[ 8.5s] [pool-1-thread-2 ] queue.put(5) returned (queue = [3, 4, 5])
[ 9.0s] [pool-1-thread-10]     Calling queue.take() (queue = [3, 4, 5])...
[ 9.0s] [pool-1-thread-10]     queue.take() returned 3 (queue = [4, 5])
[ 9.5s] [pool-1-thread-1 ] Calling queue.put(6) (queue = [4, 5])...
[ 9.5s] [pool-1-thread-1 ] queue.put(6) returned (queue = [4, 5, 6])
[10.5s] [pool-1-thread-8 ] Calling queue.put(7) (queue = [4, 5, 6])...
[11.5s] [pool-1-thread-4 ] Calling queue.put(8) (queue = [4, 5, 6])...
[12.0s] [pool-1-thread-5 ]     Calling queue.take() (queue = [4, 5, 6])...
[12.0s] [pool-1-thread-5 ]     queue.take() returned 4 (queue = [5, 6, 7])
[12.0s] [pool-1-thread-8 ] queue.put(7) returned (queue = [5, 6, 7])
[12.5s] [pool-1-thread-3 ] Calling queue.put(9) (queue = [5, 6, 7])...
[15.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [5, 6, 7])...
[15.0s] [pool-1-thread-7 ]     queue.take() returned 5 (queue = [6, 7, 8])
[15.0s] [pool-1-thread-4 ] queue.put(8) returned (queue = [6, 7, 8])
[18.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [6, 7, 8])...
[18.0s] [pool-1-thread-3 ] queue.put(9) returned (queue = [7, 8, 9])
[18.0s] [pool-1-thread-9 ]     queue.take() returned 6 (queue = [7, 8, 9])
[21.0s] [pool-1-thread-6 ]     Calling queue.take() (queue = [7, 8, 9])...
[21.0s] [pool-1-thread-6 ]     queue.take() returned 7 (queue = [8, 9])
[24.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [8, 9])...
[24.0s] [pool-1-thread-2 ]     queue.take() returned 8 (queue = [9])
[27.0s] [pool-1-thread-10]     Calling queue.take() (queue = [9])...
[27.0s] [pool-1-thread-10]     queue.take() returned 9 (queue = [])

Since we start writing only after two threads have already called take(), these first two read attempts block at 0.0 and 3.0 s (threads 1 and 4).

After 3.5 s, the first element is written (thread 8). This wakes up thread 1, and the take() method immediately removes this element from the queue again.

After 4.5 s, the second element is written (thread 5). Thread 4 is woken up and takes this element from the queue again.

The program writes faster than it reads. After 10.5 s, a writing thread (thread 8) blocks for the first time when trying to write 7 into the queue, which is full at that time. After 11.5 s, thread 4 also blocks when trying to write 8 into the queue.

After 12.0 s, thread 5 removes an element from the queue, which frees up space. Thread 8 is woken up and writes 7 into the queue.

See if you can read and understand the rest of the output yourself.

Is LinkedBlockingQueue Thread-Safe?

Yes, LinkedBlockingQueue is thread-safe.

Thread safety of LinkedBlockingQueue is guaranteed by pessimistic locking using two separate ReentrantLocks: putLock guards inserting at the end of the list, takeLock guards taking from the head. The number of elements is kept in an AtomicInteger named count, which both sides update – each under its own lock, which is why it is a non-blocking atomic addition rather than a plain counter.

So lock contention – one thread waiting for a lock the other holds – between producer and consumer threads is rare: only when a producer writes into an empty queue does it briefly take the takeLock to wake up a waiting consumer – and conversely, a consumer that reads from a full bounded queue briefly takes the putLock to wake up a waiting producer.

The following diagram shows the two locks at the two ends of the list – in the example, a queue holding the elements 3 and 4, into which a producer thread (on the right of the picture) writes 5 while a consumer thread (on the left) takes the first element:

Differences from other queues:

• With ConcurrentLinkedQueue, thread safety is provided by optimistic locking via compare-and-set, resulting in better performance with low to moderate contention.
• ArrayBlockingQueue is protected with only one ReentrantLock, so access conflicts between producer and consumer threads are possible.

LinkedBlockingQueue Time Complexity

As with all queues based on linked lists, the time required for enqueue and dequeue operations is independent of the length of the queue. The time complexity is, therefore, O(1).

That also applies to the size() method. Unlike ConcurrentLinkedQueue, which is also based on a linked list and runs through the complete list to count the elements each time size() is called, LinkedBlockingQueue uses the counter count, which is updated on insertion and removal, and thus keeps the size available in constant time.

Summary and Outlook

In this article, you have learned about LinkedBlockingQueue – a thread-safe, blocking, optionally bounded queue. It is thread-safe thanks to two locks – putLock at the end and takeLock at the head of the list – so producers and consumers rarely block each other. You saw an example of how you can use LinkedBlockingQueue, and you also learned in which cases you should use it.

LinkedBlockingQueue is based on a linked list. The next part of the tutorial is about the array-based counterpart – ArrayBlockingQueue.