ArrayBlockingQueue in Java: bounded, threadsicher, ein Lock

In diesem Artikel geht es um die ArrayBlockingQueue und deren Eigenschaften. Du siehst anhand eines Beispiels, wie die ArrayBlockingQueue eingesetzt wird. Außerdem gebe ich dir eine Empfehlung, in welchen Fällen du diese Queue einsetzen solltest.

Hier befinden wir uns in der Klassenhierarchie:

ArrayBlockingQueue Eigenschaften

Die Klasse java.util.concurrent.ArrayBlockingQueue basiert auf einem Array und ist – wie die meisten Queue-Implementierungen – threadsicher (s. u.). Sie ist bounded (hat eine feste Kapazität) und blockierend – put() wartet bei voller Queue, take() bei leerer – und bietet optional eine Fairness Policy (d. h. wartende Producer- und Consumer-Threads werden in der Reihenfolge bedient, in der sie ihre blockierende Methode aufgerufen haben).

Die Kapazität legst du beim Erzeugen fest, und sie lässt sich danach nicht mehr ändern: new ArrayBlockingQueue<>(capacity) erzeugt eine Queue ohne Fairness Policy, new ArrayBlockingQueue<>(capacity, true) eine mit. Laut Javadoc senkt Fairness in der Regel den Durchsatz, reduziert dafür die Schwankungen und verhindert, dass ein wartender Thread dauerhaft übergangen wird.

Ein dritter Konstruktor übernimmt zusätzlich die Elemente einer Collection. Eine Kapazität unter 1 quittieren alle drei Konstruktoren mit einer IllegalArgumentException.

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

Die Eigenschaften in der Übersicht:

Unterliegende Datenstruktur	Thread-safe?	Blocking/ Non-blocking	Fairness Policy	Bounded/ Unbounded	Iterator Type
Array	Ja (pessimistisches Locking mit nur einem Lock)	Blocking	Optional	Bounded	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 ArrayBlockingQueue, wenn du eine blockierende Queue mit fester Kapazität brauchst und dir Fairness oder gleichmäßige Latenzen wichtiger sind als maximaler Durchsatz – oder wenn du vermeiden willst, dass für jedes eingefügte Element ein eigenes Knotenobjekt auf dem Heap angelegt wird, das der Garbage Collector nach dem Entnehmen wieder wegräumen muss, wie das bei einer Linked Queue der Fall ist.

Die ArrayBlockingQueue legt ihr Array einmal beim Erzeugen an und schreibt die Elemente danach nur noch hinein.

Kommt es vor allem auf Durchsatz bei vielen Producern und Consumern an, ist die LinkedBlockingQueue meist schneller: Sie verwendet zum Schreiben und Lesen zwei separate ReentrantLocks, sodass Producer und Consumer sich nur selten gegenseitig blockieren. Warum das bei der ArrayBlockingQueue anders ist, zeigt der Abschnitt „Ist ArrayBlockingQueue threadsicher?“. Laut Javadoc ist die Performance von Linked Queues dafür weniger vorhersagbar. Soll gar nichts blockieren, nimm die ConcurrentLinkedQueue.

ArrayBlockingQueue Beispiel

Im folgenden Beispiel erzeugen wir eine ArrayBlockingQueue mit der Kapazität 3 und eingeschalteter Fairness Policy. Dann lassen wir über einen ScheduledExecutorService in bestimmten Abständen Elemente in die Queue schreiben und aus ihr lesen (→ Code auf GitHub):

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

  static void main() throws InterruptedException {
    BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3, true);
    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));
  }
}

Wir versuchen alle drei Sekunden, beginnend sofort, ein Element aus der Queue zu lesen. Wir schreiben die Elemente sekündlich, fangen allerdings erst nach 3,5 s damit an. Zu diesem Zeitpunkt sollten also bereits zwei lesende Threads blockiert haben und darauf warten, dass Elemente in die Queue geschrieben werden.

Da wir schneller schreiben als lesen, sollte die Queue bald ihre Kapazitätsgrenze erreicht haben. Ab dem Moment sollten die schreibenden Threads so lange blockieren, bis die lesenden Threads aufgeholt haben.

Hier eine beispielhafte Ausgabe:

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

Wie vorausgesehen blockieren die ersten zwei Leseversuche bei 0,0 s und 3,0 s, da noch keine Elemente in die Queue geschrieben wurden.

Nach 3,5 s wird das erste Element geschrieben. Dadurch wird der erste Thread aufgeweckt und entnimmt dieses Element wieder. Nach 4,5 s wird das zweite Element geschrieben und der zweite Thread aufgeweckt, um das Element zu entnehmen.

Da das Programm schneller schreibt als liest, blockieren nach 10,5 s Thread 1, nach 11,5 s Thread 9 und nach 12,5 s Thread 7 beim Versuch, weitere Elemente in die zu dem Zeitpunkt volle Queue zu schreiben.

Nach 12,0 s wird ein Element entnommen und Thread 1 kann mit dem Schreiben fortfahren. Nach 15,0 s wird ein weiteres Element entnommen und Thread 9 kann fortfahren. Nach 18,0 s kann Thread 7 fortfahren.

Da keine weiteren Elemente in die Queue geschrieben werden, leert sie sich gegen Ende wieder.

Beachte die Reihenfolge: Die drei blockierten put()-Aufrufe kommen in der Reihenfolge dran, in der sie blockiert haben. Mit eingeschalteter Fairness Policy ist diese Reihenfolge garantiert; ohne sie ist sie laut Javadoc nicht festgelegt.

Ist ArrayBlockingQueue threadsicher?

Ja, ArrayBlockingQueue ist threadsicher.

Intern nutzt die Klasse das Array als Ringpuffer: takeIndex zeigt auf das älteste Element, putIndex auf den nächsten freien Platz; erreicht einer der beiden das Ende des Arrays, springt er auf 0 zurück. Einfügen am Ende und Entnehmen am Kopf verschieben deshalb nie ein Element – nur remove(Object), das ein Element mitten aus dem Array löscht, rückt die nachfolgenden Elemente auf. Wie ein Ringpuffer Schritt für Schritt funktioniert, zeige ich im Artikel „Ringpuffer (Ring Buffer)“.

Die Threadsicherheit gewährleistet ein einzelnes ReentrantLock, das im Quelltext schlicht lock heißt. Es schützt sowohl den Kopf als auch das Ende der Queue, und an ihm hängen zwei Conditions: put() wartet auf notFull, take() auf notEmpty. Weil Producer- und Consumer-Threads dasselbe Lock nehmen, kann es zwischen ihnen auch dann zu Zugriffskonflikten („thread contention“) kommen, wenn die Queue weder leer noch voll ist.

Das folgende Diagramm zeigt das eine Lock zwischen den beiden Enden des Arrays – im Beispiel eine Queue mit der Kapazität 6 und den Elementen 3, 4 und 5, aus der ein Consumer-Thread (links im Bild) das vorderste Element entnimmt, während ein Producer-Thread (rechts im Bild) die 6 schreibt. Abgebildet ist der Moment davor: putIndex steht auf Feld 5, dem letzten Feld des Arrays, und springt nach dem Schreiben auf 0:

Bei niedriger bis moderater Contention ist optimistisches Locking – wie in der ConcurrentLinkedQueue – meist performanter; bei hoher Contention sind explizite Locks im Vorteil. Und nur sie lassen einen Thread warten, bis ein Element oder ein freier Platz da ist – eine lock-freie Queue kann nicht blockieren.

Unterschiede zu anderen Queues:

• Bei LinkedBlockingQueue wird die Threadsicherheit nicht nur durch eines, sondern durch zwei Locks gewährleistet. So blockieren Producer- und Consumer-Threads sich nur selten gegenseitig.
• 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 Zeitkomplexität

put(), take(), offer() und poll() arbeiten dank des Ringpuffers unabhängig von der Länge der Queue; ihre Zeitkomplexität beträgt O(1).

Auch size() ist O(1): Die Klasse führt den Zähler count beim Einfügen und Entnehmen mit. contains() und remove(Object) durchlaufen dagegen das Array und sind O(n).

Zusammenfassung und Ausblick

Dieser Artikel hat dir die ArrayBlockingQueue vorgestellt. Diese Queue ist threadsicher, blockierend und bounded; ein einziges ReentrantLock schützt beide Enden ihres Ringpuffers. Anhand eines Beispiels hast du gesehen, wie du sie mit eingeschalteter Fairness Policy einsetzt. Brauchst du vor allem Durchsatz, ist die LinkedBlockingQueue mit ihren zwei Locks meist die bessere Wahl.

Wie der Name schon sagt, basiert diese Queue auf einem Array. Ihr auf einer verketteten Liste basierendes Pendant, die LinkedBlockingQueue, wurde im vorherigen Teil der Serie behandelt.

Im nächsten Teil der Serie geht es um die PriorityBlockingQueue – eine threadsichere und blockierende Variante der in einem vorherigen Teil vorgestellten PriorityQueue.