LinkedBlockingQueue in Java: threadsicher mit zwei Locks

In diesem Teil der Tutorialserie geht es um die LinkedBlockingQueue. Du wirst deren spezielle Eigenschaften kennenlernen und anhand eines Beispiels sehen, wie man diese Queue einsetzt. Außerdem wirst du erfahren, wann du genau diese Queue einsetzen solltest.

Hier befinden wir uns in der Klassenhierarchie:

LinkedBlockingQueue Eigenschaften

Die Klasse java.util.concurrent.LinkedBlockingQueue basiert – genau wie ConcurrentLinkedQueue – auf einer verketteten Liste, ist allerdings – ebenso wie die im nächsten Teil vorgestellte ArrayBlockingQueue – threadsicher (s. u.), blockierend und optional bounded.

Anders als die ArrayBlockingQueue bietet LinkedBlockingQueue keine Fairness Policy an. (Fairness Policy bedeutet, dass blockierende Methoden in der Reihenfolge bedient werden, in der sie aufgerufen werden.)

Ob die Queue bounded ist, entscheidest du beim Erzeugen: new LinkedBlockingQueue<>() legt eine Queue mit der Kapazität Integer.MAX_VALUE an – praktisch unbounded. new LinkedBlockingQueue<>(capacity) setzt eine Grenze; ist sie erreicht, blockiert put(), und offer() gibt false zurück.

null-Elemente sind nicht erlaubt (offer(null) wirft eine NullPointerException) – deshalb kann poll() mit dem Rückgabewert null eindeutig „Queue leer“ signalisieren.

Die Queue-Eigenschaften im Detail:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Fairness Policy	Bounded/ Unbounded	Iterator Type
Verkettete Liste	Ja (pessimistisches Locking mit zwei Locks)	Blocking	Nicht verfügbar	Optional bounded (Default: unbounded)	Weakly consistent¹

¹ Weakly consistent: Elemente, die seit der Erzeugung des Iterators ununterbrochen in der Queue liegen, liefert der Iterator genau einmal. Änderungen nach der Erzeugung – Einfügungen wie Entnahmen – kann er berücksichtigen, muss es aber nicht; eine ConcurrentModificationException wirft er nie.

Der Iterator unterstützt remove().

Einsatzempfehlung

Ich empfehle LinkedBlockingQueue, wenn du eine blockierende, threadsichere Queue benötigst.

Gegenüber der ArrayBlockingQueue hat sie in den meisten Fällen den höheren Durchsatz, denn Producer und Consumer blockieren sich nur selten gegenseitig – wie das funktioniert, zeigt der Abschnitt „Ist LinkedBlockingQueue threadsicher?“. Dafür legt sie für jedes Element einen eigenen Knoten an, und laut Javadoc ist ihre Performance in den meisten nebenläufigen Anwendungen weniger vorhersagbar als die einer Array-basierten Queue.

Die ArrayBlockingQueue ist die bessere Wahl, wenn du eine Fairness Policy brauchst oder gleichmäßige Latenzen wichtiger sind als Durchsatz. Soll gar nichts blockieren, nimm die ConcurrentLinkedQueue.

LinkedBlockingQueue Beispiel

Das folgende Beispiel zeigt, wie die LinkedBlockingQueue verwendet wird. Wir erzeugen eine Queue mit der Kapazität 3. Direkt im Anschluss beginnen wir im Abstand von jeweils drei Sekunden Elemente aus der Queue zu lesen. Erst nach 3,5 Sekunden beginnen wir im Abstand von jeweils einer Sekunde Elemente in die Queue zu schreiben (→ Code auf GitHub).

public class LinkedBlockingQueueExample {
  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 second (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));
  }
}

Im Folgenden siehst du die Ausgabe des Beispiel-Programms:

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

Da wir erst mit dem Schreiben beginnen, nachdem bereits zwei Threads take() aufgerufen haben, blockieren diese ersten zwei Leseversuche bei 0,0 und 3,0 s (Thread 1 und 4).

Nach 3,5 s wird das erste Element geschrieben (Thread 8). Dadurch wird Thread 1 aufgeweckt und die take()-Methode entnimmt dieses Element sofort wieder aus der Queue.

Nach 4,5 s wird das zweite Element geschrieben (Thread 5). Thread 4 wird aufgeweckt und entnimmt dieses Element wieder aus der Queue.

Das Programm schreibt schneller, als es liest. Nach 10,5 s blockiert daher zum ersten Mal ein schreibender Thread (Thread 8) beim Versuch, die 7 in die – zu dem Zeitpunkt volle – Queue zu schreiben. Nach 11,5 s blockiert ebenfalls Thread 4 beim Versuch, die 8 in die Queue zu schreiben.

Nach 12,0 s entnimmt Thread 5 ein Element aus der Queue. Dadurch wird in der Queue ein Platz frei. Thread 8 wird aufgeweckt und schreibt die 7 in die Queue.

Versuch einmal selbst, die restlichen Ausgaben zu lesen und zu verstehen.

Ist LinkedBlockingQueue threadsicher?

Ja, LinkedBlockingQueue ist threadsicher.

Die Threadsicherheit der LinkedBlockingQueue wird durch pessimistisches Locking mittels zweier separater ReentrantLocks gewährleistet: putLock schützt das Einfügen am Ende der Liste, takeLock das Entnehmen am Kopf. Die Anzahl der Elemente hält ein AtomicInteger namens count, den beide Seiten aktualisieren – jede unter ihrem eigenen Lock, deshalb als nicht-blockierende atomare Addition statt als einfacher Zähler.

So kommt es zwischen Producer- und Consumer-Threads nur selten zu Lock-Contention – ein Thread wartet auf ein Lock, das der andere hält: Nur wenn ein Producer in eine leere Queue schreibt, nimmt er kurz das takeLock, um einen wartenden Consumer zu wecken – und umgekehrt nimmt ein Consumer, der aus einer vollen bounded Queue liest, kurz das putLock, um einen wartenden Producer zu wecken.

Das folgende Diagramm zeigt die beiden Locks an den zwei Enden der Liste – im Beispiel eine Queue mit den Elementen 3 und 4, in die ein Producer-Thread (rechts im Bild) die 5 schreibt, während ein Consumer-Thread (links im Bild) das vorderste Element entnimmt:

Unterschiede zu anderen Queues:

• Bei ConcurrentLinkedQueue wird die Threadsicherheit durch optimistisches Locking via Compare-and-set gewährleistet, was zu besserer Performance bei niedriger bis moderater Contention führt.
• ArrayBlockingQueue wird mit nur einem ReentrantLock geschützt, sodass dort Zugriffskonflikte zwischen Producer- und Consumer-Threads möglich sind.

LinkedBlockingQueue Zeitkomplexität

Wie bei allen Queues auf Basis verketteter Listen ist der Aufwand für die Enqueue- und Dequeue-Operationen unabhängig von der Länge der Queue. Die Zeitkomplexität beträgt also O(1).

Dies gilt auch für die size()-Methode. Im Gegensatz zur ebenfalls auf einer verketteten Liste basierenden ConcurrentLinkedQueue, die bei jedem Aufruf von size() die komplette Liste durchläuft, um die Elemente zu zählen, verwendet LinkedBlockingQueue den Zähler count, der beim Einfügen und Entnehmen aktualisiert wird, und hält so die Größe mit konstantem Aufwand verfügbar.

Zusammenfassung und Ausblick

In diesem Artikel hast du die LinkedBlockingQueue kennengelernt – eine threadsichere, blockierende, optional bounded Queue. Threadsicher ist sie durch zwei Locks – putLock am Ende und takeLock am Kopf der Liste –, sodass Producer und Consumer sich nur selten gegenseitig blockieren. An einem Beispiel hast du gesehen, wie du LinkedBlockingQueue einsetzen kannst. Du hast außerdem erfahren, in welchen Fällen du LinkedBlockingQueue einsetzen solltest.

LinkedBlockingQueue basiert auf einer verketteten Liste. Im nächsten Teil des Tutorials geht es um das auf einem Array basierende Pendant – die ArrayBlockingQueue.