Java SynchronousQueue: Direct Handoff Explained (+ Example)

This article is about a special queue – SynchronousQueue – and its properties and applications. An example will show you how to use SynchronousQueue – and why Executors.newCachedThreadPool() uses it as its work queue.

Here we are in the class hierarchy:

SynchronousQueue Characteristics

The word “Synchronous” in the java.util.concurrent.SynchronousQueue class is not to be confused with “synchronized”. Instead, it means that each enqueue operation must wait for a corresponding dequeue operation, and each dequeue operation must wait for an enqueue operation.

So every element passes directly from a producer thread to a consumer thread – a so-called handoff.

A SynchronousQueue never contains elements, even if enqueue operations are currently waiting for dequeue operations. Similarly, the size of a SynchronousQueue is always 0, and peek() always returns null.

Like all BlockingQueue implementations, SynchronousQueue does not accept null elements: put(null) throws a NullPointerException.

SynchronousQueue and ArrayBlockingQueue are the only queue implementations that offer a fairness policy. When it is enabled (new SynchronousQueue<>(true)), waiting threads are served in FIFO order. Without the fairness policy, the order is unspecified according to the documentation.

The characteristics of SynchronousQueue in detail:

Underlying data structure	Thread-safe?	Blocking/ non-blocking	Fairness policy	Bounded/ unbounded	Iterator type
Linked stack (non-fair) or linked queue (fair)	Yes (optimistic locking through compare-and-set)	Blocking	Optional	No capacity	The iterator is always empty.

When No Thread Is Waiting on the Other Side

put() and take() wait until another thread calls the counterpart operation. The other methods for inserting and removing do not wait, or wait only for a limited time.

If no thread is currently waiting on the other side, they behave as follows:

• offer(e) returns false immediately.
• poll() returns null immediately.
• offer(e, timeout, unit) and poll(timeout, unit) wait at most the specified time and then return false or null, respectively.
• add(e) throws an IllegalStateException with the message “Queue full” – a queue without capacity is always full.
• remove() and element() throw a NoSuchElementException.

If, on the other hand, a thread is currently waiting in take(), offer(e) hands the element over to it and returns true.

The thread pool in the section “SynchronousQueue in newCachedThreadPool()” relies on this behavior of offer().

Recommended Use Case

Like DelayQueue and LinkedTransferQueue, I have never used SynchronousQueue directly in my own projects.

If its characteristics fit your requirements, you can use it without hesitation. In the JDK, Executors.newCachedThreadPool() uses SynchronousQueue as its work queue, so the likelihood of bugs is extremely low.

SynchronousQueue Example

In the following example (→ code on GitHub), three threads are started that call SynchronousQueue.put(), then six threads that call SynchronousQueue.take(), and then another three threads that execute SynchronousQueue.put():

public class SynchronousQueueExample {
  private static final boolean FAIR = false;

  static void main() throws InterruptedException {
    BlockingQueue<Integer> queue = new SynchronousQueue<>(FAIR);

    // Start 3 producing threads
    for (int i = 0; i < 3; i++) {
      int element = i; // Assign to an effectively final variable
      new Thread(() -> enqueue(queue, element)).start();
      Thread.sleep(250);
    }

    // Start 6 consuming threads
    for (int i = 0; i < 6; i++) {
      new Thread(() -> dequeue(queue)).start();
      Thread.sleep(250);
    }

    // Start 3 more producing threads
    for (int i = 3; i < 6; i++) {
      int element = i; // Assign to an effectively final variable
      new Thread(() -> enqueue(queue, element)).start();
      Thread.sleep(250);
    }
  }

  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,
        "[%-9s] %s%n",
        Thread.currentThread().getName(),
        String.format(format, args));
  }
}

The output shows how the first three calls to put() (by threads 0, 1, and 2) block until the inserted elements are retrieved with take() (by threads 3, 4, and 5) in reverse order.

After that, the three following calls to take() (threads 6, 7, 8) block until three more elements have been handed over with put() (threads 9, 10, 11).

The queue remains empty for the entire time.

[Thread-0 ] Calling queue.put(0) (queue = [])...
[Thread-1 ] Calling queue.put(1) (queue = [])...
[Thread-2 ] Calling queue.put(2) (queue = [])...
[Thread-3 ]     Calling queue.take() (queue = [])...
[Thread-3 ]     queue.take() returned 2 (queue = [])
[Thread-2 ] queue.put(2) returned (queue = [])
[Thread-4 ]     Calling queue.take() (queue = [])...
[Thread-4 ]     queue.take() returned 1 (queue = [])
[Thread-1 ] queue.put(1) returned (queue = [])
[Thread-5 ]     Calling queue.take() (queue = [])...
[Thread-5 ]     queue.take() returned 0 (queue = [])
[Thread-0 ] queue.put(0) returned (queue = [])
[Thread-6 ]     Calling queue.take() (queue = [])...
[Thread-7 ]     Calling queue.take() (queue = [])...
[Thread-8 ]     Calling queue.take() (queue = [])...
[Thread-9 ] Calling queue.put(3) (queue = [])...
[Thread-9 ] queue.put(3) returned (queue = [])
[Thread-8 ]     queue.take() returned 3 (queue = [])
[Thread-10] Calling queue.put(4) (queue = [])...
[Thread-10] queue.put(4) returned (queue = [])
[Thread-7 ]     queue.take() returned 4 (queue = [])
[Thread-11] Calling queue.put(5) (queue = [])...
[Thread-11] queue.put(5) returned (queue = [])
[Thread-6 ]     queue.take() returned 5 (queue = [])

The following diagram lays out the pairs from the output. The put() calls are on the left, the take() calls on the right, each column in the order in which the calls arrived. The calls that wait have a dark blue border, the ones that come later a light blue one. Each arrow leads from a put() call to the take() call that receives its element:

The arrows cross because in both rows, the waiting thread that arrived last is served first.

If you set the FAIR constant to true, you will see the elements being taken in FIFO order rather than LIFO order. In the diagram, the arrows would then run in parallel.

SynchronousQueue in newCachedThreadPool()

The factory method Executors.newCachedThreadPool() creates a ThreadPoolExecutor with a SynchronousQueue as its work queue:

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
    60L, TimeUnit.SECONDS,
    new SynchronousQueue<Runnable>());

So the pool keeps no threads permanently (corePoolSize = 0), may start any number of them (maximumPoolSize = Integer.MAX_VALUE), and a thread without a task terminates after 60 seconds (keepAliveTime = 60 and unit = TimeUnit.SECONDS). I explain the five constructor arguments in detail in the section “PriorityBlockingQueue in a ThreadPoolExecutor”.

When a program submits a task with execute(), the ThreadPoolExecutor tries to put it into the work queue with offer(). With a SynchronousQueue, this works as follows:

1. If an idle thread of the pool is currently waiting for work in poll(), offer() hands the task over to it and returns true.
2. If no thread is waiting, offer() returns false immediately. The ThreadPoolExecutor then starts a new thread that executes the task.
3. Once a thread has finished its task, it waits for the next one with poll() for at most 60 seconds. If none arrives in that time, poll() returns null, and the thread finally terminates.

The Javadoc of ThreadPoolExecutor calls this strategy direct handoff: a task never waits in the queue but goes straight to an idle or a new thread. The pool therefore grows with the load and shrinks again when the load decreases. According to the Javadoc of Executors, such a pool typically improves the performance of programs that execute many short-lived asynchronous tasks.

This growth comes at a price: if tasks keep arriving faster than the pool can process them, it keeps starting more threads – the Javadoc warns of “unbounded thread growth”.

The pool from Executors.newFixedThreadPool() works differently: it starts a fixed number of threads and buffers all further tasks in an unbounded LinkedBlockingQueue. A task waits there until a thread becomes free. The number of threads stays fixed – in return, the queue can grow without limit.

Summary and Outlook

In this article, you learned about SynchronousQueue – a queue that never contains elements but passes them directly from producer threads to consumer threads. Without the fairness policy, it serves waiting threads in LIFO order; with the fairness policy, in FIFO order.

Executors.newCachedThreadPool() uses SynchronousQueue to hand each task over to an idle or a new thread immediately.

The next part is about the last queue implementation of this tutorial series: LinkedTransferQueue.