SynchronousQueue in Java
(+ Code Examples)

This article is about a special queue – SynchronousQueue – and its properties and applications. An example will show you how to use SynchronousQueue.

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.

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.

SynchronousQueue and ArrayBlockingQueue are the only queue implementations that offer a fairness policy. There is a peculiarity here: If the fairness policy is not activated, blocking calls are served in unspecified order according to the documentation. In fact, however, they are served precisely in reverse order (i.e., in LIFO order) since internally, SynchronousQueue uses a stack.

The characteristics of SynchronousQueue in detail:

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, SynchronousQueue is used in Executors.newCachedThreadPool() as a "work queue" for the executor, 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;

  public static void main(String[] args) 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));
  }
}
Code language: Java (java)

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 written to the queue 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 = [])Code language: plaintext (plaintext)

If you set the FAIR constant to true, you will see the elements being taken in FIFO order rather than LIFO order.

Summary and Outlook

In this article, you learned about SynchronousQueue – a queue that never contains elements but passes them directly from the enqueuing threads to the dequeuing threads.

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

If you still have questions, please ask them via the comment function. Do you want to be informed about new tutorials and articles? Then click here to sign up for the HappyCoders.eu newsletter.