LinkedBlockingDeque in Java (mit Beispiel)

In diesem Teil der Tutorialserie erfährst du alles über das LinkedBlockingDeque:

• Was sind die Eigenschaften von LinkedBlockingDeque?
• Wann sollte man es einsetzen?
• Wie setzt man es ein (Java-Beispiel)?

Wir befinden uns hier in der Klassenhierarchie:

LinkedBlockingDeque Eigenschaften

Die Klasse java.util.concurrent.LinkedBlockingDeque basiert – genau wie ConcurrentLinkedDeque – auf einer verketteten Liste, ist allerdings bounded (hat eine maximale Kapazität) und blockierend.

LinkedBlockingDeque ist das Deque-Pendant zu LinkedBlockingQueue und hat entsprechend ähnliche Eigenschaften:

• Es basiert auf einer doppelt verketteten Liste.
• Threadsicherheit wird durch ein einzelnes ReentrantLock garantiert, das von allen Enqueue- und Dequeue-Operationen geteilt wird (LinkedBlockingQueue hingegen verwendet zwei Locks – ein Enqueue-Lock und ein Dequeue-Lock).
• Im Gegensatz zu ConcurrentLinkedDeque wird die Größe des Deques in einem Feld gespeichert und nicht bei jedem Aufruf von size() durch Zählen der Listenknoten berechnet. Die Zeitkomplexität der size()-Methode lautet somit: O(1).
• LinkedBlockingDeque bietet keine Fairness Policy an, d. h. blockierende Methoden werden in undefinierter Reihenfolge bedient (mit einer Fairness Policy würden sie in der Reihenfolge bedient werden, in der sie blockiert haben).

Die Deque-Eigenschaften im Detail:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Fairness Policy	Bounded/ Unbounded	Iterator Type
Verkettete Liste	Ja (pessimistisches Locking mit einem Lock)	Blocking	Nicht verfügbar	Bounded	Weakly consistent¹

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

Einsatzempfehlung

Ich empfehle LinkedBlockingDeque, wenn du ein blockierendes, threadsicheres Deque benötigst.

Für alle anderen Einsatzzwecke schaue dir den Artikel „Deque Implementierungen – Welche einsetzen?“ an.

LinkedBlockingDeque Beispiel

Das folgende Beispiel, zeigt wie du LinkedBlockingDeque einsetzen kannst. Das Beispiel erweitert das LinkedBlockingQueue-Beispiel dahingehend, dass es Elemente auf einer zufälligen Seite des Deques einfügt bzw. entnimmt.

Folgendes passiert im Beispiel:

• Wir erstellen zunächst eine LinkedBlockingDeque mit einer Kapazität für drei Elemente.
• Dann planen wir zehn Dequeue-Operationen, die im Abstand von drei Sekunden Elemente aus dem Deque an zufälliger Seite entnehmen.
• Außerdem planen wir zehn Enqueue-Operationen, die erst nach 3,5 Sekunden starten, dann aber im Abstand von jeweils nur einer Sekunde Elemente an einer zufälligen Seite des Deques einfügen.
• Dadurch, dass wir mit den Enqueue-Operationen später starten, können wir zu Beginn blockierende Dequeue-Operationen sehen.
• Da wir dann deutlich schneller einfügen als entnehmen, erreichen wir schnell die Deque-Kapazität, so dass auch Enqueue-Threads blockieren.

Du findest den Code in der Klasse LinkedBlockingDequeExample auf GitHub.

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

  public static void main(String[] args) throws InterruptedException {
    BlockingDeque<Integer> deque = new LinkedBlockingDeque<>(3);
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);

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

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

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

  private static void enqueue(BlockingDeque<Integer> deque, int i) {
    if (ThreadLocalRandom.current().nextBoolean()) {
      enqueueAtFront(deque, i);
    } else {
      enqueueAtBack(deque, i);
    }
  }

  private static void enqueueAtFront(BlockingDeque<Integer> deque, int element) {
    log("Calling deque.putFirst(%d) (deque = %s)...", element, deque);
    try {
      deque.putFirst(element);
      log("deque.putFirst(%d) returned (deque = %s)", element, deque);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void enqueueAtBack(BlockingDeque<Integer> deque, int element) {
    log("Calling deque.putLast(%d) (deque = %s)...", element, deque);
    try {
      deque.putLast(element);
      log("deque.putLast(%d) returned (deque = %s)", element, deque);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void dequeue(BlockingDeque<Integer> deque) {
    if (ThreadLocalRandom.current().nextBoolean()) {
      dequeueAtFront(deque);
    } else {
      dequeueAtBack(deque);
    }
  }

  private static void dequeueAtFront(BlockingDeque<Integer> deque) {
    log("    Calling deque.takeFirst() (deque = %s)...", deque);
    try {
      Integer element = deque.takeFirst();
      log("    deque.takeFirst() returned %d (deque = %s)", element, deque);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private static void dequeueAtBack(BlockingDeque<Integer> deque) {
    log("    Calling deque.takeLast() (deque = %s)...", deque);
    try {
      Integer element = deque.takeLast();
      log("    deque.takeLast() returned %d (deque = %s)", element, deque);
    } 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));
  }
}

Hier siehst du eine beispielhafte Ausgabe des Programms:

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

Es ist zu Beginn gut zu sehen, wie die takeLast()- und takeFirst()-Aufrufe nach 0 s und 3 s an dem leeren Deque blockieren.

Nach 3,5 s und 4,5 s schreiben wir Elemente in das Deque, die sofort von den vorab blockierten Methoden in Threads 1 und 4 wieder entnommen werden.

Wir schreiben nun schneller als dass wir lesen, so dass nach 10,5 s Thread 1 beim Aufruf von putLast() und nach 11,5 s Thread 4 beim Aufruf von putFirst() am vollen Deque blockieren.

Nach 12 s entnimmt Thread 5 ein Element, so dass Thread 1 fortfahren und das Deque wieder füllen kann.

Nach 12,5 s blockiert Thread 9 mit putFirst(), da das Deque nach wie vor (bzw. wieder) voll ist.

Nach 15 s und 18 s entnehmen die Threads 3 und 7 jeweils ein Element, wodurch die blockierten Threads 4 und 9 wiederum ein Element einfügen können.

Danach werden (bei 21 s, 24 s und 27 s) die verbleibenden drei Elemente entnommen und keine neuen eingefügt.

Zusammenfassung und Ausblick

In diesem Teil der Tutorialserie hast du das auf einer verketteten Liste basierte, threadsichere, bounded und blockierende LinkedBlockingDeque und seine Eigenschaften kennengelernt.

Das war die letzte von vier Deque-Implementierungen. Im nächsten Teil gebe ich dir eine Entscheidungshilfe, wann du welche Deque-Implementierung einsetzten solltest.