Java ConcurrentLinkedQueue: Thread-Safe Without Locks

In this article, you will learn everything about ConcurrentLinkedQueue, its characteristics and usage scenarios. An example will show you how to use ConcurrentLinkedQueue.

Here we are in the class hierarchy:

ConcurrentLinkedQueue Characteristics

The class java.util.concurrent.ConcurrentLinkedQueue is based on a singly linked list and is – like most queue implementations – thread-safe (see below).

ConcurrentLinkedQueue is unbounded: a capacity limit would need a counter that is updated together with the list on every offer() and poll(). LinkedBlockingQueue keeps exactly such a counter; the lock-free algorithm of ConcurrentLinkedQueue does without it. ConcurrentLinkedQueue also does not provide blocking operations.

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 characteristics in detail:

Underlying data structure	Thread-safe?	Blocking/ non-blocking	Bounded/ unbounded	Iterator type
Linked list	Yes (optimistic locking via compare-and-set)	Non-blocking	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().

The individual operations offer() and poll() are atomic: each of them takes effect for other threads at exactly one moment, the successful compare-and-set. Bulk operations such as addAll(), removeIf(), and forEach() are not: a forEach() running concurrently with an addAll() may see only some of the new elements.

Recommended Use Case

ConcurrentLinkedQueue is a good choice when a thread-safe, non-blocking, unbounded queue is needed.

This applies despite my general recommendation to prefer array-based data structures over those implemented with linked lists.

Array-based alternatives are:

• The ArrayDeque described in the deque tutorial – this is, however, not thread-safe.
• The ArrayBlockingQueue described later in this tutorial – firstly, it is bounded, and secondly, it implements thread safety via a single ReentrantLock. This is, for most use cases (with low to medium contention), less performant than optimistic locking.

ConcurrentLinkedQueue Example

The following example demonstrates the thread safety of ConcurrentLinkedQueue. Four writing and three reading threads concurrently add and remove elements from the queue (→ code on GitHub):

public class ConcurrentLinkedQueueExample {

  private static final int NUMBER_OF_PRODUCERS = 4;
  private static final int NUMBER_OF_CONSUMERS = 3;
  private static final int NUMBER_OF_ELEMENTS_TO_PUT_INTO_QUEUE_PER_THREAD = 5;
  private static final int MIN_SLEEP_TIME_MILLIS = 500;
  private static final int MAX_SLEEP_TIME_MILLIS = 2000;

  private static final int POISON_PILL = -1;

  static void main() throws InterruptedException {
    Queue<Integer> queue = new ConcurrentLinkedQueue<>();

    // Start producers
    CountDownLatch producerFinishLatch = new CountDownLatch(NUMBER_OF_PRODUCERS);
    for (int i = 0; i < NUMBER_OF_PRODUCERS; i++) {
      createProducerThread(queue, producerFinishLatch).start();
    }

    // Start consumers
    for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
      createConsumerThread(queue).start();
    }

    // Wait until all producers are finished
    producerFinishLatch.await();

    // Put poison pills on the queue (one for each consumer)
    for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
      queue.offer(POISON_PILL);
    }

    // We'll let the program end when all consumers are finished
  }

  private static Thread createProducerThread(
      Queue<Integer> queue, CountDownLatch finishLatch) {
    return new Thread(
        () -> {
          ThreadLocalRandom random = ThreadLocalRandom.current();
          for (int i = 0; i < NUMBER_OF_ELEMENTS_TO_PUT_INTO_QUEUE_PER_THREAD; i++) {
            sleepRandomTime();

            Integer element = random.nextInt(1000);
            queue.offer(element);
            System.out.printf(
                "[%s] queue.offer(%3d)        --> queue = %s%n",
                Thread.currentThread().getName(), element, queue);
          }

          finishLatch.countDown();
        });
  }

  private static Thread createConsumerThread(Queue<Integer> queue) {
    return new Thread(
        () -> {
          while (true) {
            sleepRandomTime();

            Integer element = queue.poll();
            System.out.printf(
                "[%s]     queue.poll() = %4d --> queue = %s%n",
                Thread.currentThread().getName(), element, queue);

            // End the thread when a poison pill is detected
            if (element != null && element == POISON_PILL) {
              break;
            }
          }
        });
  }

  private static void sleepRandomTime() {
    ThreadLocalRandom random = ThreadLocalRandom.current();
    try {
      Thread.sleep(random.nextInt(MIN_SLEEP_TIME_MILLIS, MAX_SLEEP_TIME_MILLIS));
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
}

Here are the first 15 lines of an exemplary output:

[Thread-1] queue.offer(982)        --> queue = [982]
[Thread-6]     queue.poll() =  982 --> queue = []
[Thread-5]     queue.poll() = null --> queue = []
[Thread-0] queue.offer(917)        --> queue = [917]
[Thread-3] queue.offer(224)        --> queue = [917, 224]
[Thread-2] queue.offer(932)        --> queue = [917, 224, 932]
[Thread-6]     queue.poll() =  917 --> queue = [224, 932]
[Thread-4]     queue.poll() =  224 --> queue = [932]
[Thread-5]     queue.poll() =  932 --> queue = []
[Thread-0] queue.offer(607)        --> queue = [607]
[Thread-1] queue.offer( 87)        --> queue = [607, 87]
[Thread-3] queue.offer(264)        --> queue = [607, 87, 264]
[Thread-4]     queue.poll() =  607 --> queue = [87, 264]
[Thread-0] queue.offer(348)        --> queue = [87, 264, 348]
[Thread-2] queue.offer(728)        --> queue = [87, 264, 348, 728]

We can see very nicely how the seven threads insert and remove elements. In the third line, we see that thread 5 received null from the call to queue.poll() because the queue was empty at that time.

ConcurrentLinkedQueue Performance

This section discusses the thread safety and time complexity of ConcurrentLinkedQueue.

Is ConcurrentLinkedQueue Thread-Safe?

The thread safety of ConcurrentLinkedQueue is achieved by optimistic locking – though the word “locking” is misleading: the algorithm by Michael and Scott, on which the class is based, gets by without any lock at all. It works with non-blocking compare-and-set (CAS) operations on the next pointer of the last node when inserting and on the item of the first node when removing.

The head and tail pointers may lag behind: a thread only moves them forward if, on its way to the last or first node, it had to step over at least one node. Without contention, this happens on every second insertion or removal, which saves half of the CAS operations on these two pointers; when several threads access the queue at the same time, the gap can also become larger. A thread therefore does not rely on the two pointers but checks the node itself – next == null means last node, item != null means first node – and otherwise moves on by one node.

What happens when two threads insert at the same time is shown in the following diagram, using values from the example output: both read next = null in the last node, both try to hook their new node in there – but only one compare-and-set succeeds. The other thread re-reads next and repeats its attempt at the new end of the queue.

When accessing queues, low to moderate contention (access conflicts due to multiple threads) is usually to be expected. A thread usually does not access the queue continuously but must first create the element to be set or process the element to be taken.

With low to moderate contention, optimistic locking achieves a significant performance gain over pessimistic locking through implicit or explicit locks.

Differences from other queues:

• With LinkedBlockingQueue, thread safety is provided by pessimistic locking via two ReentrantLocks, which can lead to better performance with high contention.
• With ArrayBlockingQueue, thread safety is provided by a single ReentrantLock.

ConcurrentLinkedQueue Time Complexity

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

However, this does not apply to the size() method. To determine the length of the queue, you must iterate over all elements of the linked list. The longer the queue, the longer it takes to calculate the length. Therefore, the time complexity for size() is O(n).

And because other threads may insert and remove elements in the meantime, the result may already be outdated by the time it is returned. For the question “Is the queue empty?”, use isEmpty() instead, which only checks the head of the queue.

(Here, you can find an introduction to the topic of time complexity.)

Summary and Outlook

In this part of the tutorial series, I introduced you to the concrete Queue implementation ConcurrentLinkedQueue and its characteristics.

The following part will be about the PriorityQueue, which has some surprises in store.