Java SynchronousQueue: Handoff ohne Puffer (mit Beispiel)

In diesem Artikel geht es um eine sehr spezielle Queue – die SynchronousQueue, deren Eigenschaften und Einsatzgebiete. Anhand eines Beispiels siehst du, wie man die SynchronousQueue einsetzt – und warum Executors.newCachedThreadPool() sie als Work Queue verwendet.

Hier befinden wir uns in der Klassenhierarchie:

SynchronousQueue Eigenschaften

Das „Synchronous“ in der Klasse java.util.concurrent.SynchronousQueue ist nicht mit „synchronized“ zu verwechseln. Es bedeutet vielmehr, dass jede Enqueue-Operation auf eine korrespondierende Dequeue-Operation warten muss und jede Dequeue-Operation auf eine Enqueue-Operation.

Jedes Element geht also direkt von einem Producer-Thread an einen Consumer-Thread – ein sogenannter Handoff (Übergabe).

Eine SynchronousQueue enthält niemals Elemente, auch dann nicht, wenn Enqueue-Operationen gerade auf Dequeue-Operationen warten. Analog dazu ist die Größe einer SynchronousQueue immer 0, und peek() liefert immer null zurück.

Wie alle BlockingQueue-Implementierungen akzeptiert die SynchronousQueue keine null-Elemente: put(null) wirft eine NullPointerException.

Die SynchronousQueue und die ArrayBlockingQueue sind die einzigen Queue-Implementierungen, die eine Fairness Policy anbieten. Ist sie aktiviert (new SynchronousQueue<>(true)), werden wartende Threads in FIFO-Reihenfolge bedient. Ohne Fairness Policy ist die Reihenfolge laut Dokumentation unspezifiziert.

Die Eigenschaften der SynchronousQueue im Detail:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Fairness Policy	Bounded/ Unbounded	Iterator Type
Verketteter Stack (nicht fair) oder verkettete Queue (fair)	Ja (optimistisches Locking durch Compare-and-set)	Blocking	Optional	Keine Kapazität	Der Iterator ist immer leer.

Wenn kein Thread auf der Gegenseite wartet

put() und take() warten, bis ein anderer Thread die Gegenoperation aufruft. Die übrigen Methoden zum Einfügen und Entnehmen warten nicht oder nur begrenzt.

Wartet gerade kein Thread auf der Gegenseite, verhalten sie sich so:

• offer(e) liefert sofort false.
• poll() liefert sofort null.
• offer(e, timeout, unit) und poll(timeout, unit) warten höchstens die angegebene Zeit und liefern dann false bzw. null.
• add(e) wirft eine IllegalStateException mit der Meldung „Queue full“ – eine Queue ohne Kapazität ist immer voll.
• remove() und element() werfen eine NoSuchElementException.

Wartet dagegen gerade ein Thread in take(), übergibt offer(e) ihm das Element und liefert true.

Auf diesem Verhalten von offer() baut der Thread-Pool im Abschnitt „SynchronousQueue in newCachedThreadPool()“ auf.

Einsatzempfehlung

Genau wie die DelayQueue und die LinkedTransferQueue habe ich auch die SynchronousQueue in eigenen Projekten noch nie direkt eingesetzt.

Sollten ihre Eigenschaften zu deinen Anforderungen passen, kannst du sie bedenkenlos verwenden. Im JDK setzt Executors.newCachedThreadPool() die SynchronousQueue als Work Queue ein; die Wahrscheinlichkeit von Bugs ist also äußerst gering.

SynchronousQueue Beispiel

Im folgenden Beispiel (→ Code auf GitHub) werden zunächst drei Threads gestartet, die SynchronousQueue.put() aufrufen, danach sechs Threads, die SynchronousQueue.take() aufrufen und anschließend noch einmal drei Threads, die SynchronousQueue.put() ausführen:

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));
  }
}

In der Ausgabe kannst du sehen, wie die ersten drei Aufrufe von put() durch die Threads 0, 1 und 2 so lange blockieren, bis die eingefügten Elemente mit take() durch die Threads 3, 4 und 5 in umgekehrter Reihenfolge entnommen werden.

Danach blockieren die drei folgenden Aufrufe von take() (Threads 6, 7, 8) so lange, bis mit put() drei weitere Elemente übergeben wurden (Threads 9, 10, 11).

Die Queue bleibt während der gesamten Zeit leer.

[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 = [])

Das folgende Diagramm stellt die Paare aus der Ausgabe gegenüber. Links stehen die put()-Aufrufe, rechts die take()-Aufrufe, jede Spalte in der Reihenfolge, in der die Aufrufe ankamen. Dunkelblau umrandet sind die Aufrufe, die warten, hellblau die, die danach kommen. Ein Pfeil führt jeweils vom put()-Aufruf zu dem take()-Aufruf, der sein Element erhält:

Die Pfeile kreuzen sich, weil in beiden Reihen der zuletzt angekommene wartende Thread als Erster bedient wird.

Wenn du die Konstante FAIR auf true setzt, wirst du sehen, wie die Elemente nicht in LIFO-, sondern in FIFO-Reihenfolge entnommen werden. Im Diagramm liefen die Pfeile dann parallel.

SynchronousQueue in newCachedThreadPool()

Die Factory-Methode Executors.newCachedThreadPool() erzeugt einen ThreadPoolExecutor mit einer SynchronousQueue als Work Queue:

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

Der Pool hält also keine Threads dauerhaft (corePoolSize = 0), darf beliebig viele starten (maximumPoolSize = Integer.MAX_VALUE), und ein Thread ohne Aufgabe beendet sich nach 60 Sekunden (keepAliveTime = 60 und unit = TimeUnit.SECONDS). Was die fünf Argumente des Konstruktors bedeuten, erkläre ich ausführlich im Abschnitt „PriorityBlockingQueue im ThreadPoolExecutor“.

Reicht ein Programm mit execute() eine Aufgabe ein, versucht der ThreadPoolExecutor, sie mit offer() in die Work Queue zu legen. Mit einer SynchronousQueue läuft das so ab:

1. Wartet ein freier Thread des Pools gerade in poll() auf Arbeit, übergibt offer() ihm die Aufgabe und liefert true.
2. Wartet kein Thread, liefert offer() sofort false. Der ThreadPoolExecutor startet dann einen neuen Thread, der die Aufgabe ausführt.
3. Hat ein Thread seine Aufgabe erledigt, wartet er mit poll() höchstens 60 Sekunden auf die nächste. Kommt in dieser Zeit keine, liefert poll() null, und der Thread beendet sich schließlich.

Die Javadoc von ThreadPoolExecutor nennt diese Strategie Direct Handoff (direkte Übergabe): Eine Aufgabe wartet nie in der Queue, sondern geht sofort an einen freien oder einen neuen Thread. Der Pool wächst also mit der Last und schrumpft wieder, wenn sie nachlässt. Laut Javadoc von Executors verbessert ein solcher Pool typischerweise die Performance von Programmen mit vielen kurzlebigen asynchronen Aufgaben.

Dieses Wachstum hat einen Preis: Kommen Aufgaben dauerhaft schneller herein, als der Pool sie abarbeitet, startet er immer weitere Threads – die Javadoc warnt vor „unbounded thread growth“.

Anders arbeitet der Pool aus Executors.newFixedThreadPool(): Er startet eine feste Anzahl Threads und puffert alle weiteren Aufgaben in einer unbounded LinkedBlockingQueue. Dort wartet eine Aufgabe, bis ein Thread frei wird. Die Zahl der Threads bleibt fest – dafür kann die Queue unbegrenzt wachsen.

Zusammenfassung und Ausblick

In diesem Artikel hast du die SynchronousQueue kennengelernt – eine Queue, die niemals Elemente enthält, sondern diese direkt von Producer-Threads an Consumer-Threads übergibt. Ohne Fairness Policy bedient sie wartende Threads in LIFO-Reihenfolge, mit Fairness Policy in FIFO-Reihenfolge.

Executors.newCachedThreadPool() nutzt die SynchronousQueue, um jede Aufgabe sofort an einen freien oder einen neuen Thread zu übergeben.

Im nächsten Teil geht es um die letzte Queue-Implementierung dieser Tutorial-Serie: die LinkedTransferQueue.