BlockingQueue-Interface in Java (mit Beispiel)

In diesem Artikel lernst du das Interface java.util.concurrent.BlockingQueue kennen. BlockingQueue erweitert das im vorherigen Teil dieser Tutorial-Serie besprochene Queue-Interface um Methoden zum blockierenden Zugriff auf Queues.

Bevor wir klären, was „blockierender Zugriff“ bedeutet, müssen wir zunächst über den Begriff „bounded Queue“ sprechen.

Was ist eine bounded Queue?

Wenn eine Queue nur eine begrenzte Anzahl von Elementen aufnehmen kann, spricht man von einer „bounded Queue“. Die maximale Anzahl von Elementen wird mit „Kapazität“ (englisch „capacity“) bezeichnet und beim Erzeugen der Queue einmalig festgelegt.

Die folgende Code-Zeile beispielsweise erzeugt eine auf 100 Elemente begrenzte ArrayBlockingQueue:

Queue<Integer> queue = new ArrayBlockingQueue<>(100);

Ist die Anzahl der Elemente in der Queue hingegen nicht begrenzt (bzw. nur durch den zur Verfügung stehenden Speicher), spricht man von einer „unbounded Queue“.

(Dieselbe Definition gilt übrigens für alle Datenstrukturen, z. B. auch für Stacks und Deques.)

Was ist eine blockierende Queue?

Bei den Queue-Operationen „Enqueue“ und „Dequeue“ können zwei Sonderfälle auftreten:

• Wir könnten versuchen ein Element in eine bounded Queue einzufügen, die ihre Kapazitätsgrenze erreicht hat – oder anders gesagt: die voll ist.
• Wir könnten versuchen ein Element aus einer leeren Queue zu entnehmen.

Eine nicht blockierende („non-blocking“) Queue liefert in solchen Fällen einen bestimmten Rückgabewert oder wirft eine Exception (s. Abschnitt „Queue-Methoden“ im Artikel über Java-Queues).

Eine blockierende Queue hingegen bietet zusätzliche Methoden, die darauf warten, dass die gewünschte Operation ausgeführt werden kann:

• Enqueue-Methoden, die beim Einfügen in eine volle bounded Queue warten, bis diese wieder freie Kapazitäten hat (dazu muss ein anderer Thread ein Element entnehmen).
• Dequeue-Methoden, die beim Entnehmen eines Elements aus einer leeren Queue darauf warten, dass diese nicht mehr leer ist (dazu muss ein anderer Thread ein Element einfügen).

Diese zusätzlichen Methoden sind im BlockingQueue-Interface definiert und werden im folgenden Kapitel erläutert.

BlockingQueue Interface

Die blockierenden Enqueue- und Dequeue-Operationen gibt es in jeweils zwei Varianten. Die erste Variante wartet unbegrenzt lange. Die zweite Variante gibt nach Ablauf einer vorgegebenen Wartezeit auf und liefert false bzw. null zurück.

Die folgende Tabelle zeigt in den ersten zwei Spalten die nicht blockierenden Methoden, die BlockingQueue von Queue erbt (und die wir im vorherigen Teil des Tutorials besprochen haben). In der dritten und vierten Spalte findest du die hinzugekommenen, blockierenden Methoden:

Operation	Nicht blockierend: Exception	Nicht blockierend: Rückgabewert	Blockierend	Blockierend mit Timeout
Element anhängen (enqueue)	add(e)	offer(e)	put(e)	offer(e, time, unit)
Element entnehmen (dequeue)	remove()	poll()	take()	poll(time, unit)
Element ansehen (examine)	element()	peek()	–	–

Der folgende Abschnitt beschreibt die BlockingQueue-Methoden im Einzelnen.

BlockingQueue Methoden

BlockingQueue.put()

Die put()-Methode fügt, sofern Platz vorhanden ist, ein Element in die Queue ein. Ist die Kapazitätsgrenze der Queue hingegen erreicht, blockiert die Methode so lange, bis Platz frei geworden ist.

BlockingQueue.offer() mit Timeout

Auch die offer()-Methode fügt ein Element ein, wenn in der Queue noch Platz ist. Andernfalls wartet die Methode die angegebene Zeit. Wird in dieser Zeit ein Platz frei, wird das Element eingefügt und die Methode gibt true zurück. Läuft die Wartezeit hingegen ab, ohne dass ein Platz frei geworden ist, gibt die Methode false zurück.

BlockingQueue.take()

Diese Methode entnimmt ein Element vom Kopf der Queue, sofern die Queue nicht leer ist. Bei einer leeren Queue blockiert take() so lange, bis ein Element verfügbar geworden ist, und gibt dieses dann zurück.

BlockingQueue.poll() mit Timeout

Auch poll() entnimmt ein Element vom Kopf der Queue, wenn diese nicht leer ist. Ist die Queue hingegen leer, wartet die Methode die angegebene Zeit. Sofern in der Wartezeit ein Element verfügbar wird, wird dieses entnommen und zurückgegeben. Läuft die Wartezeit ergebnislos ab, gibt die Methode null zurück.

BlockingQueue.remainingCapacity()

Diese Methode gibt zurück, wie viele Elemente die Queue noch aufnehmen kann, ohne dass das Einfügen blockiert. Eine unbounded Queue gibt immer Integer.MAX_VALUE zurück.

Ob das nächste Einfügen ohne Warten gelingt, verrät remainingCapacity() allerdings nicht zuverlässig: Zwischen dem Aufruf und dem Einfügen kann ein anderer Thread bereits ein Element eingefügt oder entnommen haben.

BlockingQueue.drainTo()

drainTo(c) entnimmt alle Elemente, die gerade in der Queue liegen, und fügt sie in die Collection c ein. drainTo(c, maxElements) entnimmt höchstens maxElements Elemente. Beide Methoden geben die Anzahl der übertragenen Elemente zurück. Sie warten nicht darauf, dass Elemente verfügbar werden. Laut Dokumentation können sie effizienter sein als wiederholte poll()-Aufrufe.

InterruptedException bei blockierenden Methoden

Alle blockierenden Methoden werfen eine InterruptedException, wenn auf dem wartenden Thread die interrupt()-Methode aufgerufen wird. Mit interrupt() sollten wartende Threads abgebrochen werden, wenn das Warten nicht mehr nötig ist.

Dies ist z. B. beim Herunterfahren der Applikation der Fall. Dabei wird eventuell das Ereignis, auf das die blockierende Methode wartet, nicht mehr eintreten. Die Methode würde aber dennoch auf den Eintritt des Ereignisses warten und damit ein reguläres Herunterfahren der Applikation verhindern. Das Abbrechen der wartenden Threads mit interrupt() ermöglicht ein sauberes Herunterfahren.

Bevor eine blockierende Methode die InterruptedException wirft, setzt sie den Interrupt-Status des Threads zurück. Wer die Exception fängt und nicht weiterwirft, sollte den Status deshalb mit Thread.currentThread().interrupt() wiederherstellen – so wie es das folgende Beispiel in seinen catch-Blöcken tut.

Weitere Eigenschaften einer BlockingQueue

Für alle BlockingQueue-Implementierungen gilt außerdem:

• Keine null-Elemente: Eine BlockingQueue nimmt kein null auf; add(), put() und offer() werfen in diesem Fall eine NullPointerException. Denn null ist der Rückgabewert, mit dem poll() meldet, dass es kein Element gab.
• Thread-Sicherheit: Alle Queue-Methoden wirken atomar. Für die Massenoperationen aus Collection – addAll(), containsAll(), retainAll() und removeAll() – gilt das nicht zwingend: addAll() kann z. B. mit einer Exception abbrechen, nachdem es bereits einen Teil der Elemente eingefügt hat.
• Sichtbarkeit: Alles, was ein Thread vor dem Einfügen eines Elements tut, ist für den Thread sichtbar, der dieses Element anschließend entnimmt. Im Java Memory Model heißt das: Das Einfügen happens-before dem Entnehmen.
• Virtuelle Threads: Wartet ein virtueller Thread in put() oder take(), blockiert er keinen Plattform-Thread. Die Implementierungen des JDK warten über LockSupport.park(), und ein geparkter virtueller Thread gibt seinen Carrier-Thread frei.

Java BlockingQueue Beispiel

Der folgende Quellcode zeigt ein Beispiel, welches aufgrund der Nebenläufigkeit deutlich komplexer ausfällt als das Beispiel mit einer nicht-blockierenden Queue (→ Code in GitHub):

public class BlockingQueueExample {
  private static final long startTime = System.currentTimeMillis();

  static void main() throws InterruptedException {
    BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(3);
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);

    // Start reading from the queue immediately, every 3 seconds
    for (int i = 0; i < 10; i++) {
      int delaySeconds = i * 3;
      pool.schedule(() -> dequeue(queue), delaySeconds, TimeUnit.SECONDS);
    }

    // Start writing to the queue after 3.5 seconds (so there are already 2
    // threads waiting), every 1 seconds (so that the queue fills faster than
    // it's emptied, so that we see a full queue soon)
    for (int i = 0; i < 10; i++) {
      int element = i; // Assign to an effectively final variable
      int delayMillis = 3500 + i * 1000;
      pool.schedule(
          () -> enqueue(queue, element), delayMillis, TimeUnit.MILLISECONDS);
    }

    pool.shutdown();
    pool.awaitTermination(1, TimeUnit.MINUTES);
  }

  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,
        "[%4.1fs] [%-16s] %s%n",
        (System.currentTimeMillis() - startTime) / 1000.0,
        Thread.currentThread().getName(),
        String.format(format, args));
  }
}

In diesem Beispiel erstellen wir eine blocking, bounded Queue mit der Kapazität 3 und schedulen jeweils zehn Enqueue- und zehn Dequeue-Operationen.

Die Enqueue-Operationen beginnen später, so dass wir am Anfang blockierende Dequeue-Operationen sehen können. Außerdem erfolgen die Enqueue-Operationen in kürzeren Abständen, so dass die Kapazitätsgrenze der Queue nach einer Weile erreicht ist und wir blockierende Enqueue-Operationen sehen können.

Hier die Ausgabe des Programms:

[ 0.0s] [pool-1-thread-1 ]     Calling queue.take() (queue = [])...
[ 3.0s] [pool-1-thread-3 ]     Calling queue.take() (queue = [])...
[ 3.5s] [pool-1-thread-2 ] Calling queue.put(0) (queue = [])...
[ 3.5s] [pool-1-thread-2 ] queue.put(0) returned (queue = [])
[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 0 (queue = [])
[ 4.5s] [pool-1-thread-10] Calling queue.put(1) (queue = [])...
[ 4.5s] [pool-1-thread-10] queue.put(1) returned (queue = [])
[ 4.5s] [pool-1-thread-3 ]     queue.take() returned 1 (queue = [])
[ 5.5s] [pool-1-thread-8 ] Calling queue.put(2) (queue = [])...
[ 5.5s] [pool-1-thread-8 ] queue.put(2) returned (queue = [2])
[ 6.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [2])...
[ 6.0s] [pool-1-thread-9 ]     queue.take() returned 2 (queue = [])
[ 6.5s] [pool-1-thread-5 ] Calling queue.put(3) (queue = [])...
[ 6.5s] [pool-1-thread-5 ] queue.put(3) returned (queue = [3])
[ 7.5s] [pool-1-thread-6 ] Calling queue.put(4) (queue = [3])...
[ 7.5s] [pool-1-thread-6 ] queue.put(4) returned (queue = [3, 4])
[ 8.5s] [pool-1-thread-7 ] Calling queue.put(5) (queue = [3, 4])...
[ 8.5s] [pool-1-thread-7 ] queue.put(5) returned (queue = [3, 4, 5])
[ 9.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [3, 4, 5])...
[ 9.0s] [pool-1-thread-4 ]     queue.take() returned 3 (queue = [4, 5])
[ 9.5s] [pool-1-thread-2 ] Calling queue.put(6) (queue = [4, 5])...
[ 9.5s] [pool-1-thread-2 ] queue.put(6) returned (queue = [4, 5, 6])
[10.5s] [pool-1-thread-1 ] Calling queue.put(7) (queue = [4, 5, 6])...
[11.5s] [pool-1-thread-10] Calling queue.put(8) (queue = [4, 5, 6])...
[12.0s] [pool-1-thread-3 ]     Calling queue.take() (queue = [4, 5, 6])...
[12.0s] [pool-1-thread-3 ]     queue.take() returned 4 (queue = [5, 6, 7])
[12.0s] [pool-1-thread-1 ] queue.put(7) returned (queue = [5, 6, 7])
[12.5s] [pool-1-thread-8 ] Calling queue.put(9) (queue = [5, 6, 7])...
[15.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [5, 6, 7])...
[15.0s] [pool-1-thread-9 ]     queue.take() returned 5 (queue = [6, 7, 8])
[15.0s] [pool-1-thread-10] queue.put(8) returned (queue = [6, 7, 8])
[18.0s] [pool-1-thread-5 ]     Calling queue.take() (queue = [6, 7, 8])...
[18.0s] [pool-1-thread-5 ]     queue.take() returned 6 (queue = [7, 8, 9])
[18.0s] [pool-1-thread-8 ] queue.put(9) returned (queue = [7, 8, 9])
[21.0s] [pool-1-thread-6 ]     Calling queue.take() (queue = [7, 8, 9])...
[21.0s] [pool-1-thread-6 ]     queue.take() returned 7 (queue = [8, 9])
[24.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [8, 9])...
[24.0s] [pool-1-thread-7 ]     queue.take() returned 8 (queue = [9])
[27.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [9])...
[27.0s] [pool-1-thread-4 ]     queue.take() returned 9 (queue = [])

Die folgende Zeitleiste fasst die Ausgabe zusammen. Oben siehst du, wie viele Elemente in der Queue liegen, darunter die Aufrufe, die warten mussten:

Zu Beginn ist die Queue leer, so dass die ersten zwei Leseversuche (nach 0 und 3 s) blockieren.

Nach 3,5 s (nachdem also zwei lesende Threads an der Queue warten) beginnt das Programm sekündlich in die Queue zu schreiben. Man sieht in der Ausgabe schön, wie dabei jeweils ein lesender Thread fortgesetzt wird und das angehängte Element sofort wieder entnimmt (bei 3,5 und 4,5 s).

Da das Programm dreimal so schnell in die Queue schreibt, wie es daraus liest, blockiert nach 10,5 s der Versuch, eine 7 in die Queue zu schreiben: Die Queue hat mit den Elementen [4, 5, 6] ihre Kapazitätsgrenze von 3 erreicht.

Erst nachdem nach 12 s die 4 aus der Queue entnommen wurde, kann die 7 eingefügt werden. Für die 8 und die 9 sehen wir ein entsprechendes Verhalten.

BlockingQueue Implementierungen

Im JDK gibt es fünf Klassen, die das BlockingQueue-Interface direkt implementieren, jeweils mit spezifischen Eigenschaften. Zwei weitere Klassen implementieren es über ein Sub-Interface: LinkedTransferQueue über TransferQueue und LinkedBlockingDeque über BlockingDeque. In dem folgenden UML-Klassendiagramm sind die fünf direkten Implementierungen mitsamt ihrem Interface farbig hinterlegt:

Die Implementierungen werden jeweils in separaten Teilen des Tutorials behandelt. Dort werden deren Eigenschaften vorgestellt und anhand dieser erläutert, unter welchen Voraussetzungen die jeweilige Implementierung eingesetzt werden sollte. Folgende Links führen zu den entsprechenden Artikeln:

• LinkedBlockingQueue
• ArrayBlockingQueue
• PriorityBlockingQueue
• DelayQueue
• SynchronousQueue
• LinkedTransferQueue über TransferQueue
• LinkedBlockingDeque über BlockingDeque

Du gelangst zu diesen Artikeln auch jederzeit über die Tutorial-Navigation am rechten Rand.

Fairness Policy

Wartende Threads werden nicht automatisch in der Reihenfolge bedient, in der sie ihre blockierenden Methoden aufgerufen haben. Bei ArrayBlockingQueue und SynchronousQueue lässt sich die Abarbeitung in Aufrufreihenfolge über den Konstruktorparameter fair aktivieren.

Diese „Fairness Policy“ verringert im Allgemeinen den Durchsatz der Queue, sorgt aber für gleichmäßigere Wartezeiten und verhindert, dass ein Thread „verhungert“. In der Regel ist die Aktivierung der „Fairness Policy“ nicht nötig.

Zusammenfassung und Ausblick

Dieser Artikel hat zunächst die Unterschiede zwischen bounded/unbounded und blocking/non-blocking Queues erläutert. Im Anschluss hast du das BlockingQueue-Interface kennengelernt: die blockierenden Methoden put(), offer(), take() und poll(), dazu remainingCapacity() und drainTo() sowie die Eigenschaften, die für alle Implementierungen gelten.

In den folgenden Teilen dieser Serie werden wir alle Queue- und BlockingQueue-Implementierungen und deren Eigenschaften genauer unter die Lupe nehmen.