LinkedTransferQueue und TransferQueue in Java (mit Beispiel)

In diesem Artikel lernst du eine sehr spezielle Queue kennen: die LinkedTransferQueue. Dieser Artikel beschreibt deren Eigenschaften und zeigt dir anhand eines Beispiels, wie du diese Queue einsetzt.

Wir befinden uns nunmehr am untersten Punkt der Queue-Klassenhierarchie:

TransferQueue Interface

Wie im Klassendiagramm gut zu erkennen ist, ist java.util.concurrent.LinkedTransferQueue die einzige Klasse, die das Interface TransferQueue implementiert.

TransferQueue definiert zusätzliche Enqueue-Methoden, die nur dann erfolgreich ausgeführt werden können, wenn ein anderer Thread mit take() oder poll() das übergebene Element entgegennimmt:

• transfer(E e) – übermittelt das Element an einen Thread, der in take() oder einem zeitbegrenzten poll() auf ein Element wartet. Wenn solch ein Thread nicht existiert, blockiert die Methode so lange, bis ein anderer Thread take() oder poll() aufruft. Das Element liegt dabei schon in der Queue – size(), peek() und der Iterator sehen es –, nur der aufrufende Thread wartet, bis es entnommen wurde.
• tryTransfer(E e) – übermittelt das Element an einen Thread, der in take() oder einem zeitbegrenzten poll() auf ein Element wartet. Wenn solch ein Thread nicht existiert, gibt die Methode umgehend false zurück.
• tryTransfer(E e, long timeout, TimeUnit unit) – übermittelt das Element an einen Thread, der in take() oder einem zeitbegrenzten poll() auf ein Element wartet. Wenn solch ein Thread nicht existiert und auch nicht innerhalb der Wartezeit erscheint, gibt die Methode false zurück.

Außerdem kannst du mit hasWaitingConsumer() und getWaitingConsumerCount() abfragen, ob bzw. wie viele Threads gerade in take() oder einem zeitbegrenzten poll() auf ein Element warten.

LinkedTransferQueue Eigenschaften

Bei der LinkedTransferQueue handelt es sich um eine unbounded blocking Queue, d. h. die regulären Enqueue-Operationen put() und offer() können nicht blockieren (da die Queue beliebig groß werden kann). Blockieren können hingegen:

• die Dequeue-Operationen (wenn die Queue leer ist)
• und die transfer()-Methode sowie die zeitbegrenzte tryTransfer()-Variante des TransferQueue-Interfaces, bis die jeweiligen Elemente entnommen werden.

LinkedTransferQueue basiert auf einer einfach verketteten Liste. Das hat zur Folge, dass die Zeitkomplexität der size()-Methode O(n) beträgt¹ (und nicht O(1) wie bei den Array-basierten Queues), da zum Bestimmen der Länge die komplette Liste abgelaufen werden muss.

Threadsicherheit wird durch nicht-blockierende Compare-and-set (CAS)-Operationen erreicht, was eine hohe Performance bei niedriger bis moderater Contention (Zugriffskonflikten durch mehrere Threads) sicherstellt.

Die Eigenschaften im Detail:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Fairness Policy	Bounded/ Unbounded	Iterator Type
Verkettete Liste	Ja (optimistisches Locking durch Compare-and-set)	Blocking	Nicht verfügbar	Unbounded	Weakly consistent²

¹ Alles über Zeitkomplexität erfährst du im Artikel „O-Notation und Zeitkomplexität – anschaulich erklärt“

² Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Iterators in der Queue liegen, werden vom Iterator genau einmal durchlaufen. Änderungen, die danach erfolgen, können – müssen aber nicht – durch den Iterator berücksichtigt werden.

Einsatzempfehlung

Lange Zeit war die LinkedTransferQueue eine Randfigur des JDK: Die Klasse entstand 2009 zusammen mit dem Fork/Join-Framework und diente während dessen Entwicklung als Submission-Queue des ForkJoinPool. Noch vor dem Release von Java 7 ersetzte das Framework sie jedoch durch eigene Work-Queues. Seit Java 22 ist das anders: Die SynchronousQueue ist seitdem intern eine LinkedTransferQueue, und damit läuft jeder mit Executors.newCachedThreadPool() erzeugte Thread-Pool auf eben dieser Queue.

Die Klasse ist also erprobt und wird aktiv gepflegt – zuletzt behob Java 26 einen Fehler, bei dem poll() unter hoher Contention null zurückgab, obwohl die Queue nicht leer war (JDK-8371740).

Einsetzen solltest du die LinkedTransferQueue, wenn ein Producer wissen muss, dass sein Element abgenommen wurde – etwa beim Nachrichtenaustausch zwischen Threads mit Empfangsbestätigung –, und du trotzdem die Möglichkeit brauchst, andere Elemente mit put() ohne Warten einzureihen. Brauchst du nur das eine oder nur das andere, sind SynchronousQueue bzw. LinkedBlockingQueue die einfachere Wahl.

LinkedTransferQueue Beispiel

Im folgenden Beispiel (→ Code auf GitHub) werden zwei Threads gestartet, die LinkedTransferQueue.transfer() aufrufen. Danach wird ein Element direkt in die Queue geschrieben. Dann werden zwei weitere Threads gestartet, die transfer() aufrufen. Abschließend werden so lange Elemente aus der Queue entnommen, bis diese wieder leer ist.

public class LinkedTransferQueueExample {
  static void main() throws InterruptedException {
    TransferQueue<Integer> queue = new LinkedTransferQueue<>();

    // Start 2 threads calling queue.transfer(),
    startTransferThread(queue, 1);
    startTransferThread(queue, 2);

    // ... then put one element directly,
    enqueueViaPut(queue, 3);

    // ... then start 2 more threads calling queue.transfer().
    startTransferThread(queue, 4);
    startTransferThread(queue, 5);

    // Now take all elements until the queue is empty
    while (!queue.isEmpty()) {
      dequeueViaTake(queue);
    }
  }

  private static void startTransferThread(TransferQueue<Integer> queue, int element)
      throws InterruptedException {
    new Thread(() -> enqueueViaTransfer(queue, element)).start();

    // Wait a bit to let the thread enqueue the element
    Thread.sleep(100);
    log("                            --> queue = " + queue);
  }

  private static void enqueueViaTransfer(TransferQueue<Integer> queue, int element) {
    log("Calling queue.transfer(%d)...", element);
    try {
      queue.transfer(element);
      log("queue.transfer(%d) returned  --> queue = %s", element, queue);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void enqueueViaPut(TransferQueue<Integer> queue, int element)
      throws InterruptedException {
    log("Calling queue.put(%d)...", element);
    queue.put(element);
    log("queue.put(%d) returned       --> queue = %s", element, queue);
  }

  private static void dequeueViaTake(TransferQueue<Integer> queue)
      throws InterruptedException {
    log("    Calling queue.take() (queue = %s)...", queue);
    Integer e = queue.take();
    log("    queue.take() returned %d --> queue = %s", e, queue);

    // Wait a bit to get the log output in a readable order
    Thread.sleep(10);
  }

  private static void log(String format, Object... args) {
    System.out.printf(
        Locale.US, "[%-8s] %s%n",
        Thread.currentThread().getName(),
        String.format(format, args));
  }
}

Im Folgenden siehst du die Ausgabe des Programms (die Zeilen von main und den Transfer-Threads können von Lauf zu Lauf die Plätze tauschen):

[Thread-0] Calling queue.transfer(1)...
[main    ]                             --> queue = [1]
[Thread-1] Calling queue.transfer(2)...
[main    ]                             --> queue = [1, 2]
[main    ] Calling queue.put(3)...
[main    ] queue.put(3) returned       --> queue = [1, 2, 3]
[Thread-2] Calling queue.transfer(4)...
[main    ]                             --> queue = [1, 2, 3, 4]
[Thread-3] Calling queue.transfer(5)...
[main    ]                             --> queue = [1, 2, 3, 4, 5]
[main    ]     Calling queue.take() (queue = [1, 2, 3, 4, 5])...
[main    ]     queue.take() returned 1 --> queue = [2, 3, 4, 5]
[Thread-0] queue.transfer(1) returned  --> queue = [2, 3, 4, 5]
[main    ]     Calling queue.take() (queue = [2, 3, 4, 5])...
[main    ]     queue.take() returned 2 --> queue = [3, 4, 5]
[Thread-1] queue.transfer(2) returned  --> queue = [3, 4, 5]
[main    ]     Calling queue.take() (queue = [3, 4, 5])...
[main    ]     queue.take() returned 3 --> queue = [4, 5]
[main    ]     Calling queue.take() (queue = [4, 5])...
[main    ]     queue.take() returned 4 --> queue = [5]
[Thread-2] queue.transfer(4) returned  --> queue = [5]
[main    ]     Calling queue.take() (queue = [5])...
[main    ]     queue.take() returned 5 --> queue = []
[Thread-3] queue.transfer(5) returned  --> queue = []

Das folgende Diagramm zeigt den Ablauf im Zeitverlauf – eine Zeile pro Element, die Zeit läuft von links nach rechts:

Zu Beginn wird zweimal transfer() aufgerufen – beide Aufrufe kehren nicht zurück, obwohl die Elemente bereits in der Queue liegen (queue = [1], dann queue = [1, 2]). Dann wird einmal put() aufgerufen, und dieser Aufruf kehrt sofort zurück. Danach folgen zwei weitere transfer()-Aufrufe, die ebenfalls blockieren.

Sobald main das erste Element entnimmt, kehrt transfer(1) zurück; mit der Entnahme der 2 kehrt transfer(2) zurück. Die Entnahme der 3 löst keine weitere Aktion aus, da diese mit put() in die Queue geschrieben wurde. Nach der Entnahme der 4 und der 5 kehren schließlich transfer(4) und transfer(5) zurück.

Zusammenfassung und Ausblick

In diesem Artikel hast du das TransferQueue-Interface und die LinkedTransferQueue-Implementierung kennengelernt und an einem Beispiel gesehen, wie du sie einsetzt.

Einen Vergleich aller Queue-Implementierungen des JDK und eine Übersicht, in welchen Fällen du welche Implementierung verwenden solltest, findest du im Artikel über das Queue-Interface.

Im nächsten Teil dieser Tutorialserie zeige ich dir, wie du Queues selbst implementieren kannst, beginnend mit der Implementierung mit einem Stack.