Kategorie: Algorithmen und Datenstrukturen

Im vorherigen Teil der Tutorialserie haben wir eine Queue mit einem Array implementiert. In diesem letzten Teil der Serie zeige ich dir, wie du eine Priority Queue implementierst – und zwar mit Hilfe eines Heaps.

Zur Erinnerung: Bei einer Priority Queue werden die Elemente nicht in FIFO-Reihenfolge entnommen, sondern entsprechend ihrer Priorität. Das Element mit der höchsten Priorität steht immer am Kopf der Queue und wird zuerst entnommen – unabhängig davon, wann es in die Queue eingefügt wurde.

Was ist ein Heap?

Ein „Heap“ ist einen Binärbaum, in dem jeder Knoten entweder größer/gleich seiner Kinder ist („Max Heap“) – oder kleiner/gleich seiner Kinder („Min-Heap“).

Für die Priority Queue in diesem Artikel verwenden wir einen Min-Heap, da die höchste Priorität diejenige mit der niedrigsten Zahl ist (Prio 1 ist in der Regel höher als Prio 2).

Hier ist ein Beispiel, wie so ein Min-Heap aussehen könnte:

Das Element an jedem Knoten dieses Baums ist kleiner als die Elemente seiner beiden Kindknoten:

• 1 ist kleiner als 2 und 4;
• 2 ist kleiner als 3 und 7;
• 4 ist kleiner als 9 und 6;
• 3 ist kleiner als 8 und 5.

Array-Darstellung eines Heaps

Ein Heap können wir in einem Array speichern, indem wir dessen Elemente zeilenweise – von links oben nach rechts unten – auf das Array abbilden:

Unser Beispiel-Heap sieht als Array wie folgt aus:

Bei einem Min-Heap ist das kleinste Element immer oben, im Array also immer an erster Position. Genau deshalb siehst du, wenn du eine Java-PriorityQueue als String ausgibst, immer das kleinste Element links. Was du siehst, ist die Array-Darstellung des der PriorityQueue zugrunde liegenden Min-Heaps.

Mit den folgenden Codezeilen lässt sich das gut demonstrieren:

PriorityQueue<Integer> priorityQueue = new PriorityQueue<>();
priorityQueue.addAll(List.of(4, 7, 3, 8, 2, 9, 6, 5, 1));
System.out.println("priorityQueue = " + priorityQueue);
Code-Sprache: Java (java)

Das Ausgabe des Programms lautet:

priorityQueue = [1, 2, 4, 3, 7, 9, 6, 8, 5]
Code-Sprache: Klartext (plaintext)

Das kleinste Element ist ganz links. Und wenn du genau hinschaust, erkennst du, dass die Zahlen exakt die gleiche Reihenfolge haben wie in der grafischen Array-Darstellung oben. Der Min-Heap der im Beispiel erstellten PriorityQueue ist genau der, den ich zu Beginn des Artikels abgebildet habe.

Priority Queue mit einem Min-Heap – Der Algorithmus

OK, das kleinste Element steht immer ganz links. Damit ist schon mal klar, wie die peek()-Operation zu funktionieren hat: sie muss einfach das erste Element des Arrays zurückgeben.

Doch wie wird so ein Heap aufgebaut? Wie funktionieren enqueue() und dequeue()?

Einfügen in den Min-Heap: Sift Up

Um ein Element in einen Heap einzufügen, gehen wir wie folgt vor:

1. Wir fügen das neue Element als letztes Element in den Baums ein, d. h.:
   • Ist der Baum leer, fügen wir das neue Element als Wurzel ein.
   • Wenn die unterste Ebene des Baums nicht voll ist, fügen wir das neue Element neben dem letzen Knoten der untersten Ebene ein.
   • Ist die unterste Ebene voll, dann hängen wir den Knoten unter den ersten Knoten der untersten Ebene.
2. Solange der Elternknoten des neuen Elements kleiner ist als das Element selbst (was die Min-Heap Regel verletzten würde), vertauschen wir den neuen Knoten mit seinem Elternknoten.

Schritt 1 hört sich kompliziert an, bedeutet in der Array-Darstellung aber lediglich, dass das neue Element an die erste freie Stelle des Arrays gesetzt wird. Schritt 2 sorgt dafür, dass am Ende der Operation wieder jedes Element kleiner ist als seine Kinder.

Das Beispiel im folgenden Abschnitt demonstriert die beiden Schritte.

Einfügen in den Min-Heap: Beispiel

An den folgenden Beispielen zeige ich dir Schritt für Schritt, wie eine auf einem Min-Heap basierende Priority Queue mit den oben gezeigten Beispielwerten (4, 7, 3, 8, 2, 9, 6, 5, 1) gefüllt wird. In jedem Schritt zeige ich den Min-Heap jeweils in Baum- und in Array-Darstellung.

1. Element – Einfügen der 4 in eine leere Priority Queue

Das erste einzufügende Element wird zum Wurzelknoten des Baumes; im Array wird es an die erste Position gesetzt:

2. Element – Einfügen der 7

Die 7 wird unter den ersten Knoten der untersten Ebene eingehängt – also links unter die Wurzel. Im Array wird sie einfach angehängt:

Die 7 ist größer als ihr Elternknoten 4 – damit ist die Einfügeoperation abgeschlossen. Das kleinste Element steht nach wie vor am Anfang der Priority Queue.

3. Element – Einfügen der 3

Die 3 wird neben dem letzten Knoten der untersten Ebene, also rechts unter der 4, angehängt. Im Array kommt sie ans Ende:

Die 3 ist kleiner als ihr Elternknoten. Die Regeln des Min-Heaps sind also verletzt. Wir stellen den Min-Heap wieder her, in dem wir die 3 mit der 4 tauschen:

Damit haben wir wieder einen validen Min-Heap.

Wir überspringen die 8, 2, 9, 6 und 5 (diese werden analog eingefügt) und kommen zum…

9. Element – Einfügen der 1

Zuletzt fügen wir die 1 am Ende der Queue (und des Arrays) ein:

Die 1 ist größer als ihr Elternknoten 5; unser Baum ist also kein gültiger Min-Heap mehr. Um ihn zu reparieren, tauschen wir zunächst die 1 mit der 5:

Die 1 ist ebenfalls größer als ihr neuer Elternknoten 3; somit tauschen wir erneut:

Die 1 ist auch größer als die Wurzel 2, also tauschen wir ein drittes Mal:

Da die 1 nun an der Wurzel angelangt ist, ist die Operation beendet. Der Baum ist wieder ein Min-Heap. Das kleinste Element steht an der Wurzel des Baumes (und am Anfang des Arrays).

Dieses Hochreichen des eingefügten Elements auf die eben gezeigte Weise bezeichnet man als „Sift Up“.

Vereinfachter Sift-Up-Algorithmus

Tatsächlich brauchen wir uns gar nicht die Mühe zu machen das neue Element am Ende einzufügen, um es dann schrittweise mit seinem Elternknoten zu tauschen. Stattdessen können wir uns das neue Element merken, die größeren Elternelemente nach unten schieben und zum Schluss das neue Element direkt an seine Zielposition setzen.

Die folgenden Grafiken zeigen das Einfügen der 1 nach dem vereinfachten Algorithmus.

Die 1 ist kleiner als der Elternknoten des freien Knotens. Wir schieben daher die 5 in den freien Knoten:

Die einzufügende 1 ist ebenso kleiner als die 3; wir schieben die 3 nach unten:

Die 1 ist kleiner als die 2; wir schieben auch die 2 nach unten:

Wir können keine weiteren Elemente nach unten schieben und setzen das einzufügende Element, die 1, auf den freigewordenen Wurzelknoten (oder das erste Feld des Arrays):

Damit ist die Sift-Up-Operation abgeschlossen.

Das Einfügen eines Elements in die Priority Queue (bzw. in den Min-Heap) mag beim ersten Durchlesen sehr komplex erscheinen. Falls du es nicht verstanden hast, mach eine kurze Pause und wiederhole das Kapitel noch einmal, bevor du zur Dequeue-Operation fortschreitest.

Entnahme aus dem Min-Heap: Sift Down

Wir wissen, dass das kleinste Element immer an der Wurzel des Baumes steht (oder am Anfang des Arrays).

Um es zu entnehmen, gehen wir wie folgt vor:

1. Wir entfernen das Wurzelelement aus dem Baum.
2. Wir verschieben den letzten Knoten der letzten Ebene des Baumes (das entspricht dem letzten Element des Arrays) an die freigewordene Wurzelposition.
3. Solange dieser Knoten größer ist als eines seiner Kinder (was die Min-Heap Regel verletzten würde), vertauschen wir den Knoten mit seinem kleinsten Kindknoten.

Entnahme aus dem Min-Heap: Beispiel

Das folgende Beispiel zeigt, wie wir das Wurzelelement des im letzten Kapitel gefüllten Min-Heaps entnehmen und danach die Min-Heap-Bedingung wiederherstellen.

Als erstes entnehmen wir das Wurzelelement:

Als zweites verschieben wir das letzte Element des Baumes, die 5, an die frei gewordene Wurzel:

Da das neue Wurzelelement, die 5, jetzt größer ist als das kleinste seiner Kinder, die 2, vertauschen wir diese beiden Elemente:

Die 5 ist weiterhin größer als das kleinste ihrer Kinder, die 3. Wir tauschen ein zweites Mal:

Die 5 ist jetzt größer als ihr einziges Kind; die Min-Heap-Bedingung ist damit wiederhergestellt.

An der Wurzel des Min-Heaps (bzw. am Anfang des Arrays) befindet sich jetzt die 2, das neue kleinste Element nach Entnahme der 1.

Das Herunterwandern des an die Wurzel verschobenen Elements bezeichnet man als „Sift Down“.

Vereinfachter Sift-Down-Algorithmus

Auch den Sift-Down-Algorithmus können wir vereinfachen. Wir müssen das letzte Element (im Beispiel die 5) nicht erst an die Wurzel schieben, um es dann schrittweise mit seinen Kindern zu tauschen. Wir können auch zuerst die größeren Elemente nach oben schieben und am Ende das letzte Element direkt an seine endgültige Position verschieben.

Die folgenden Grafiken zeigen das Herunterreichen der 5 (oder besser gesagt: des freien Feldes, auf das am Ende die 5 gesetzt wird) nach dem vereinfachten Algorithmus.

Die 5 ist größer als der kleinste Kindknoten der leeren Wurzel, die 2. Wir schieben die 2 nach oben:

Die 5 ist ebenso größer als das kleinste Kind des jetzt freien Platzes, die 3. Wir schieben auch die 3 nach oben:

Die 5 ist nicht größer als das einzige Kind des jetzt freien Platzes, die 8. Wir haben also das Zielfeld für die 5 gefunden und schieben die 5 dorthin:

Wir haben die Min-Heap-Bedingung wiederhergestellt.

Die Sift-Up- und Sift-Down-Operationen mögen komplex erscheinen, sie können allerdings beide mit maximal 10 Zeilen Java-Code implementiert werden. Wie, erfährst du im nächsten Kapitel.

Quellcode für Priority Queue mit Min-Heap

Der folgende Quellcode zeigt, wie man eine Priority Queue mit einem Min-Heap implementiert (Klasse HeapPriorityQueue im GitHub-Repository). Aufgrund der Länge der Klasse zeige ich sie im Folgenden abschnittsweise.

Konstruktoren

Es gibt zwei Konstrukturen: einen, bei dem man die initiale Größe des Arrays angeben kann und einen Default-Konstruktor, der die initiale Kapazität auf zehn setzt:

public class HeapPriorityQueue<E extends Comparable<? super E>> implements Queue<E> {

  private static final int DEFAULT_INITIAL_CAPACITY = 10;
  private static final int ROOT_INDEX = 0;

  private Object[] elements;
  private int numberOfElements;

  public HeapPriorityQueue() {
    this(DEFAULT_INITIAL_CAPACITY);
  }

  public HeapPriorityQueue(int capacity) {
    if (capacity < 1) {
      throw new IllegalArgumentException("Capacity must be 1 or higher");
    }

    elements = new Object[capacity];
  }
Code-Sprache: Java (java)

enqueue()

Die enqueue()-Methode prüft zunächst, ob das Array voll ist. Ist das der Fall, ruft sie die grow()-Methode auf, welche das Array in ein neues, größeres Array kopiert:

  @Override
  public void enqueue(E newElement) {
    if (numberOfElements == elements.length) {
      grow();
    }
    siftUp(newElement);
    numberOfElements++;
  }

  private void grow() {
    int newCapacity = elements.length + elements.length / 2;
    elements = Arrays.copyOf(elements, newCapacity);
  }Code-Sprache: Java (java)

Die grow()-Methode habe ich hier stark vereinfacht dargestellt, da der Fokus auf den siftUp()– und siftDown()-Methoden liegen soll.

In der HeapPriorityQueue-Klasse im GitHub-Repository vergrößert die grow()-Methode das Array bis zu einer bestimmten Größe (64 Elemente) um Faktor 2 und erst danach um Faktor 1,5. Außerdem wird dort sichergestellt, dass eine bestimmte Maximalgröße nicht überschritten wird.

Wenn wir sichergestellt haben, dass das Array groß genug ist, rufen wir die siftUp()-Methode auf:

siftUp()

  private void siftUp(E newElement) {
    int insertIndex = numberOfElements;

    while (isNotRoot(insertIndex) && isParentGreater(insertIndex, newElement)) {
      copyParentDownTo(insertIndex);
      insertIndex = parentOf(insertIndex);
    }

    elements[insertIndex] = newElement;
  }

  private boolean isNotRoot(int index) {
    return index != ROOT_INDEX;
  }

  private boolean isParentGreater(int insertIndex, E element) {
    int parentIndex = parentOf(insertIndex);
    E parent = elementAt(parentIndex);
    return parent.compareTo(element) > 0;
  }

  private void copyParentDownTo(int insertIndex) {
    int parentIndex = parentOf(insertIndex);
    elements[insertIndex] = elements[parentIndex];
  }

  private int parentOf(int index) {
    return (index - 1) / 2;
  }
Code-Sprache: Java (java)

Beachte, dass ich versucht habe den Algorithmus so verständlich wie möglich (und nicht so performant wie möglich) zu implementieren. Die parentOf()-Methode wird in jeder Iteration drei Mal aufgerufen: einmal von isParentGreater(), einmal von copyParentDownTo() und einmal direkt.

Eine verbesserten Variante (Klasse OptimizedHeapPriorityQueue im GitHub-Repo, ab Zeile 74) zeigt einen optimierten Algorithmus, der den Elternindex nur ein Mal berechnet.

dequeue()

Die dequeue()-Methode entnimmt das Kopf-Element, entfernt das letzte Element und ruft dann siftDown() auf, das letztendlich dieses letzte Element an seine neue Position verschiebt.

  @Override
  public E dequeue() {
    E result = elementAtHead();
    E lastElement = removeLastElement();
    siftDown(lastElement);
    return result;
  }

  private E removeLastElement() {
    numberOfElements--;
    E lastElement = elementAt(numberOfElements);
    elements[numberOfElements] = null;
    return lastElement;
  }
Code-Sprache: Java (java)

siftDown()

siftDown() ist die komplizierteste Methode, da sie einen Knoten immer mit ggf. zwei Kindknoten vergleichen muss.

  private void siftDown(E lastElement) {
    int lastElementInsertIndex = ROOT_INDEX;
    while (isGreaterThanAnyChild(lastElement, lastElementInsertIndex)) {
      moveSmallestChildUpTo(lastElementInsertIndex);
      lastElementInsertIndex = smallestChildOf(lastElementInsertIndex);
    }

    elements[lastElementInsertIndex] = lastElement;
  }

  private boolean isGreaterThanAnyChild(E element, int parentIndex) {
    E leftChild = leftChildOf(parentIndex);
    E rightChild = rightChildOf(parentIndex);

    return leftChild != null && element.compareTo(leftChild) > 0
        || rightChild != null && element.compareTo(rightChild) > 0;
  }

  private E leftChildOf(int parentIndex) {
    int leftChildIndex = leftChildIndexOf(parentIndex);
    return exists(leftChildIndex) ? elementAt(leftChildIndex) : null;
  }

  private int leftChildIndexOf(int parentIndex) {
    return 2 * parentIndex + 1;
  }

  private E rightChildOf(int parentIndex) {
    int rightChildIndex = rightChildIndexOf(parentIndex);
    return exists(rightChildIndex) ? elementAt(rightChildIndex) : null;
  }

  private int rightChildIndexOf(int parentIndex) {
    return 2 * parentIndex + 2;
  }

  private boolean exists(int index) {
    return index < numberOfElements;
  }

  private void moveSmallestChildUpTo(int parentIndex) {
    int smallestChildIndex = smallestChildOf(parentIndex);
    elements[parentIndex] = elements[smallestChildIndex];
  }

  private int smallestChildOf(int parentIndex) {
    int leftChildIndex = leftChildIndexOf(parentIndex);
    int rightChildIndex = rightChildIndexOf(parentIndex);

    if (!exists(rightChildIndex)) {
      return leftChildIndex;
    }

    return smallerOf(leftChildIndex, rightChildIndex);
  }

  private int smallerOf(int leftChildIndex, int rightChildIndex) {
    E leftChild = elementAt(leftChildIndex);
    E rightChild = elementAt(rightChildIndex);
    return leftChild.compareTo(rightChild) < 0 ? leftChildIndex : rightChildIndex;
  }
Code-Sprache: Java (java)

Genau wie siftUp() habe ich auch siftDown() mit Fokus auf Lesbarkeit geschrieben, nicht auf Performance. So werden hier pro Iteration drei Mal die Positionen der Kindelemente berechnet: in isGreaterThanAnyChild(), in moveSmallestChildUpTo() und noch einmal in smallestChildOf().

In der optimierten Klasse OptimizedHeapPriorityQueue werden diese Positionen nur einmal berechnet. Dadurch ist der Code aber auch nicht mehr ganz so leicht zu lesen.

peek(), isEmpty() und zwei Hilfsmethoden

Und zum Schluss noch die peek()– und isEmpty()-Methoden sowie zwei Hilfsmethoden, die verwendet werden, um das Element vom Kopf der Queue oder von einer spezifischen Position zu lesen.

Da wir die Elemente in einem Object-Array speichern, müssen die Array-Elemente nach E gecastet werden. Damit die Casts nicht über den gesamten Quellcode verteilt sind, habe ich das Casten in eine zentrale Stelle, die Methode elementAt(), ausgelagert und dort die „unchecked“-Warnung einmalig unterdrückt.

  @Override
  public E peek() {
    return elementAtHead();
  }

  private E elementAtHead() {
    E element = elementAt(0);
    if (element == null) {
      throw new NoSuchElementException();
    }
    return element;
  }

  private E elementAt(int child) {
    @SuppressWarnings("unchecked")
    E element = (E) elements[child];
    return element;
  }

  @Override
  public boolean isEmpty() {
    return numberOfElements == 0;
  }
}Code-Sprache: Java (java)

Wenn dir der Kopf noch nicht raucht, schau dir auch gerne einmal den Quellcode der PriorityQueue-Klasse des JDK an. Diese kann Elemente nicht nur nach ihrer natürlichen Ordnung sortieren, sondern auch anhand eines dem Konstruktor übergebenen Comparators.

Fazit

Damit endet diese Tutorial-Serie über Queues. In dieser Serie hast du gelernt, wie eine Queue funktioniert, was bounded und unbounded, blocking und non-blocking Queues sind, welche Queue-Implementierungen es im JDK gibt und wie man Queues auf verschiedene Arten selbst implementieren kann.

Wenn dir die Serie gefallen hat, hinterlasse mir gerne einen Kommantar oder teile die Artikel über die Share-Buttons am Ende. Wenn du noch Fragen hast, kannst du sie gerne über die Kommentar-Funktion stellen.

Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

Im letzten Teil der Tutorialserie ging es darum, wie man eine Queue mit einer verketteten Liste implementiert. In diesem Teil implementieren wir eine Queue mit einem Array – zunächst eine bounded Queue (also eine mit fester Kapazität) – und dann eine unbounded Queue (also eine, deren Kapazität sich ändern kann).

Beginnen wir mit der einfachen Variante, der bounded Queue.

Implementierung einer bounded Queue mit einem Array

Wir legen ein leeres Array an und füllen dieses von links nach rechts (also von Index 0 an aufsteigend) mit den in die Queue eingefügten Elementen.

Die folgende Grafik zeigt eine Queue mit einem Array namens elements, das acht Elemente fassen kann. Bisher wurden sechs Elemente in die Queue eingefügt. Der tailIndex zeigt immer die nächste Einfügeposition:

Beim Entnehmen der Elemente lesen wir diese ebenfalls von links nach rechts aus und entfernen sie aus dem Array. Der headIndex zeigt dabei immer die nächste Leseposition.

Die folgende Darstellung zeigt die Queue, nachdem wir vier der sechs Elemente wieder entnommen haben:

Da wir nun kurz vor dem Ende des Arrays angekommen sind, könnten wir (ohne zusätzliche Logik) nur noch zwei Elemente in die Queue schreiben. Um die Queue wieder auf acht Elemente aufzufüllen, gibt es zwei Lösungsmöglichkeiten:

1. Wir schieben die verbleibenden Elemente nach links, an den Anfang des Arrays. Diese Operation ist, insbesondere bei großen Arrays, sehr teuer.
2. Die bessere Lösung ist ein Ringbuffer (englisch: „circular array“). Das bedeutet, dass wir, wenn wir am Ende des Arrays angekommen sind, am Anfang des Arrays fortfahren. Das gilt sowohl für die Schreib- als auch die Leseoperation.

Ringbuffer

Um die Funktionsweise des Ringbuffers zu verdeutlichen, habe ich das Array aus dem Beispiel kreisförmig dargestellt:

Weitere Elemente fügen wir im Uhrzeigersinn ein, im folgenden Beispiel „mango“, „fig“, „pomelo“ und „apricot“ an die Positionen 6, 7 – und dann 0 und 1:

Zurück in der „flachen“ Darstellung sieht das Array nunmehr wie folgt aus:

Sowohl in der Kreisdarstellung als auch in dieser ist gut zu sehen, dass nach dem Element „fig“ an Index 7 das Element „pomelo“ an Index 0 folgt.

Das Entnehmen der Elemente erfolgt analog. Mit jeder Dequeue-Operation wandert der headIndex um eine Position nach rechts, wobei auf die 7 nicht die 8, sondern die 0 folgt.

Volle Queue vs. leere Queue

tailIndex und headIndex liegen sowohl bei einer leeren als auch bei einer vollen Queue auf derselben Position. Damit wir erkennen können, wann die Queue voll ist, speichern wir zusätzlich die Anzahl der Elemente.

So könnte eine volle Queue aussehen:

Und so eine leere (z. B. nachdem aus der eben dargestellten Queue alle acht Elemente entnommen wurden):

Das Speichern der Anzahl der Elemente ist nicht die einzige Möglichkeit, aber eine sehr simple, um eine volle von einer leeren Queue zu unterscheiden. Alternativen sind z. B.:

• Man speichert (neben der Anzahl der Elemente) nur den tailIndex oder den headIndex und berechnet den jeweils anderen aus der Anzahl der Elemente (das macht den Code aber deutlich komplexer).
• Man speichert die Anzahl der Elemente nicht und erkennt eine volle Queue daran, dass tailIndex und headIndex gleich sind und dass das Array an der Position tailIndex kein Element enthält.
• Man füllt die Queue nicht komplett, sondern lässt immer mindestens ein Feld frei.

Quellcode für die bounded Queue mit einem Array

Die Implementierung einer bounded Queue mit einem Array ist recht einfach. Du findest den folgenden Code auch in der Klasse BoundedArrayQueue im GitHub-Repository.

public class BoundedArrayQueue<E> implements Queue<E> {

  private final Object[] elements;
  private int headIndex;
  private int tailIndex;
  private int numberOfElements;

  public BoundedArrayQueue(int capacity) {
    if (capacity < 1) {
      throw new IllegalArgumentException("Capacity must be 1 or higher");
    }

    elements = new Object[capacity];
  }

  @Override
  public void enqueue(E element) {
    if (numberOfElements == elements.length) {
      throw new IllegalStateException("The queue is full");
    }
    elements[tailIndex] = element;
    tailIndex++;
    if (tailIndex == elements.length) {
      tailIndex = 0;
    }
    numberOfElements++;
  }

  @Override
  public E dequeue() {
    final E element = elementAtHead();
    elements[headIndex] = null;
    headIndex++;
    if (headIndex == elements.length) {
      headIndex = 0;
    }
    numberOfElements--;
    return element;
  }

  @Override
  public E peek() {
    return elementAtHead();
  }

  private E elementAtHead() {
    if (isEmpty()) {
      throw new NoSuchElementException();
    }

    @SuppressWarnings("unchecked")
    E element = (E) elements[headIndex];

    return element;
  }

  @Override
  public boolean isEmpty() {
    return numberOfElements == 0;
  }
}
Code-Sprache: Java (java)

Beachte, dass BoundedArrayQueue nicht das Interface java.util.Queue implementiert, sondern ein eigenes, das nur die vier Methoden enqueue(), dequeue(), peek() und isEmpty() definiert (s. Queue im GitHub-Repository):

public interface Queue<E> {
  void enqueue(E element);
  E dequeue();
  E peek();
  boolean isEmpty();
}Code-Sprache: Java (java)

Wie du die BoundedArrayQueue (und alle anderen Implementierungen des Queue-Interfaces) benutzt, kannst du dir im Programm QueueDemo ansehen.

Implementierung einer unbounded Queue mit einem Array

Etwas komplexer wird die Implementierung für eine unbounded Queue, also eine Queue ohne Größenbeschränkung. Ein Array kann ja nicht wachsen. Und selbst wenn – es dürfte nicht am Ende wachsen, sondern müsste an genau der Stelle freien Platz erzeugen, an der sich tailIndex und headIndex befinden.

Schauen wir uns noch einmal die volle Queue vom Ende des vorherigen Beispiels an:

Um ein weiteres Element einzufügen, müssen wir die Kapazität der Queue erhöhen, indem wir das Array vergrößern.

(Aus Platzgründen in der grafischen Darstellung erhöhen wir die Kapazität nur um zwei Elemente. In der Realität findet man Erweiterungen um Faktor 1,5 oder 2,0.)

Wir müssten diesen freien Platz allerdings zwischen Ende und Anfang der Queue, also innerhalb des Arrays schaffen:

Das ist nicht ohne weiteres möglich. Ein Array kann nicht wachsen – schon gar nicht in der Mitte. Stattdessen müssen wir ein neues Array erzeugen und die bestehenden Elemente dorthinein kopieren.

Wenn wir die Elemente aber ohnehin umkopieren müssen, können wir sie dabei auch in der richtigen Reihenfolge an den Anfang des neuen Arrays kopieren, z. B. so:

Der Code dafür ist gar nicht so kompliziert, wie du im nächsten Abschnitt sehen wirst.

Quellcode für die unbounded Queue mit einem Array

Der folgende Code zeigt die Klasse ArrayQueue aus dem Tutorial-GitHub-Repository.

Es gibt zwei Konstrukturen: einen, bei dem man die initiale Größe des Arrays angeben kann und einen Default-Konstruktor, der die initiale Kapazität auf zehn setzt.

Bei jedem Aufruf der enqueue()-Methode wird geprüft, ob das Array voll ist. Wenn das der Fall ist, wird die grow()-Methode aufgerufen.

Diese ruft zunächst calculateNewCapacity() auf, um die neue Größe des Arrays zu berechnen. Ich habe diese Methode hier in vereinfachter Form wiedergegeben: Sie multipliziert die aktuelle Größe mit 1,5.

Die calculateNewCapacity()-Methode im GitHub-Repository hat einen ausgefeilteren Algorithmus und stellt sicher, dass eine bestimmte Maximalgröße nicht überschritten wird. Der Fokus dieses Artikels soll jedoch nicht auf der Bestimmung der neuen Größe liegen, sondern auf der eigentlichen Erweiterung des Arrays.

Dazu wird in der Methode growToNewCapacity() das neue Array angelegt, die Elemente an die entsprechenden Positionen im neuen Array kopiert und headIndex sowie tailIndex neu gesetzt.

public class ArrayQueue<E> implements Queue<E> {

  private static final int DEFAULT_INITIAL_CAPACITY = 10;

  private Object[] elements;
  private int headIndex;
  private int tailIndex;
  private int numberOfElements;

  public ArrayQueue() {
    this(DEFAULT_INITIAL_CAPACITY);
  }

  public ArrayQueue(int capacity) {
    if (capacity < 1) {
      throw new IllegalArgumentException("Capacity must be 1 or higher");
    }

    elements = new Object[capacity];
  }

  @Override
  public void enqueue(E element) {
    if (numberOfElements == elements.length) {
      grow();
    }
    elements[tailIndex] = element;
    tailIndex++;
    if (tailIndex == elements.length) {
      tailIndex = 0;
    }
    numberOfElements++;
  }

  private void grow() {
    int newCapacity = calculateNewCapacity(elements.length);
    growToNewCapacity(newCapacity);
  }

  static int calculateNewCapacity(int currentCapacity) {
    return currentCapacity + currentCapacity / 2;
  }

  private void growToNewCapacity(int newCapacity) {
    Object[] newArray = new Object[newCapacity];

    // Copy to the beginning of the new array: tailIndex to end of the old array
    int oldArrayLength = elements.length;
    int numberOfElementsAfterTail = oldArrayLength - tailIndex;
    System.arraycopy(elements, tailIndex, newArray, 0, numberOfElementsAfterTail);

    // Append to the new array: beginning to tailIndex of the old array
    if (tailIndex > 0) {
      System.arraycopy(elements, 0, newArray, numberOfElementsAfterTail, tailIndex);
    }

    // Adjust head and tail
    headIndex = 0;
    tailIndex = oldArrayLength;
    elements = newArray;
  }

  // dequeue(), peek(), elementAtHead(), isEmpty() are the same as in BoundedArrayQueue

}
Code-Sprache: Java (java)

Die Methoden dequeue(), peek(), elementAtHead() und isEmpty() gleichen denen in der BoundedArrayQueue aus dem vorletzten Abschnitt. Ich habe sie daher nicht noch einmal mit abgedruckt.

Vielleicht ist dir aufgefallen, dass die Queue zwar wachsen, aber nicht wieder schrumpfen kann. Evtl. muss unsere Queue nur bei Lastspitzen eine hohen Zahl von Elementen speichern und würde danach den Speicher mit einem größtenteils leeren Array belegen. Wir könnten die Queue dahingehend erweitern, dass sie nach einer gewissen Karenzzeit ihren Inhalt wieder in ein kleineres Array umkopiert.

Diese Erweiterung überlasse ich dir als Übungsaufgabe.

Ausblick

Im nächsten und gleichzeitig letzten Teil dieser Tutorialserie zeige ich dir, wie man eine PriorityQueue selbst implementiert, und zwar auf Basis eines Min-Heaps.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

Im letzten Teil dieser Tutorialserie habe ich dir gezeigt, wie man eine Queue mit Stacks implementiert. In diesem Teil werden wir eine Queue mit einer verketteten Liste implementieren.

Der Algorithmus – Schritt für Schritt

Unsere Queue besteht aus zwei Referenzen auf Listenknoten: einer head– und einer tail-Referenz.

Die head-Referenz zeigt auf einen Listenknoten, der das vorderste Element der Queue enthält sowie einen next-Zeiger auf einen zweiten Listenknoten. Dieser wiederum enthält das zweite Element und einen Zeiger auf den dritten Listenknoten, usw.

Der letzte Knoten wird sowohl vom next-Zeiger des vorletzten Elements als auch vom tail-Zeiger referenziert. Er enthält das letzte Queue-Element, und dessen next-Referenz zeigt auf null.

Die folgende Grafik zeigt eine Beispiel-Queue, in die die Elemente „banana“, „cherry“ und „grape“ (in dieser Reihenfolge) eingefügt wurden:

Wie gelangen wir diesen Zustand?

Enqueue-Algorithmus

Wir beginnen mit einer leeren Queue. Sowohl head– aus auch tail-Referenz sind null.

Das erste Element fügen wir in die Queue ein, indem wir es in einen Listenknoten wrappen und sowohl head als auch tail auf diesen Knoten zeigen lassen:

Weitere Elemente fügen wir wie folgt ein:

1. Wir wrappen das einzufügende Element in einem neuen Listenknoten.
2. Wir lassen den next-Zeiger des letzten Knotens, also tail.next, auf den neuen Knoten zeigen.
3. Wir lassen ebenso tail auf den neuen Knoten zeigen.

In der folgenden Grafik siehst du, wie ein zweites Element, „cherry“, in die Beispiel-Queue eingefügt wird:

Dequeue-Algorithmus

Die Entnahme des Kopf-Elements mit dequeue() funktioniert dann wie folgt:

1. Wir merken uns das Element des Knotens, auf den head zeigt (im Beispiel wäre das „banana“).
2. Wir lassen head auf head.next zeigen (im Beispiel auf den Knoten, der „cherry“ wrappt). Falls head danach null sein sollte (die Queue also leer ist), setzen wir auch tail auf null.
3. Wir geben das in Schritt 1 gemerkte Element zurück (im Beispiel „banana“).
4. In einer Programmiersprache mit Garbage Collector (z. B. Java) löscht dieser den nicht mehr referenzierten Knoten; in anderen Sprachen (wie C++) müssten wir ihn manuell löschen.

Die folgende Grafik soll die vier Schritte verdeutlichen:

Der gestrichelte Rahmen um den „banana“-Knoten in Schritt 2 und 3 soll darstellen, dass dieser Knoten zu diesem Zeitpunkt nicht mehr referenziert wird.

Quellcode für die Queue mit einer Linked List

Der folgende Code zeigt die Implementierung einer Queue mit einer verketteten Liste (LinkedListQueue im GitHub-Repo). Die Klasse für die Knoten, Node, findest du ganz am Ende als statische innere Klasse.

public class LinkedListQueue<E> implements Queue<E> {

  private Node<E> head;
  private Node<E> tail;

  @Override
  public void enqueue(E element) {
    Node<E> newNode = new Node<>(element);
    if (isEmpty()) {
      head = tail = newNode;
    } else {
      tail.next = newNode;
      tail = newNode;
    }
  }

  @Override
  public E dequeue() {
    if (isEmpty()) {
      throw new NoSuchElementException();
    }
    E element = head.element;
    head = head.next;
    if (head == null) {
      tail = null;
    }
    return element;
  }

  @Override
  public E peek() {
    if (isEmpty()) {
      throw new NoSuchElementException();
    }
    return head.element;
  }

  @Override
  public boolean isEmpty() {
    return head == null;
  }

  private static class Node<E> {
    final E element;
    Node<E> next;

    Node(E element) {
      this.element = element;
    }
  }
}
Code-Sprache: Java (java)

Wie du die LinkedListQueue-Klasse einsetzen kannst, kannst du dir Demo-Programm QueueDemo anschauen.

Ausblick

Im nächsten Teil des Tutorials zeige ich dir, wie man eine Queue mit einem Array implementiert.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Teil der Tutorialserie zeige ich dir, wie man eine Queue mit einem Stack (genauer gesagt: mit zwei Stacks) implementiert.

Diese Variante hat keinen praktischen Nutzen, sondern ist in erster Linie eine Übungsaufgabe. Als solche ist sie das Gegenstück zur Implementierung eines Stacks mit einer Queue.

Zur Erinnerung: ein Stack ist eine Datenstruktur, bei der die Elemente in der umgekehrten Reihenfolge des Einfügens abgerufen werden, also eine Last-in-first-out-Datenstruktur (LIFO).

Wie können wir damit eine Queue, also eine First-in-first-out-Datenstruktur (FIFO) implementieren?

Die Lösung – Schritt für Schritt

Das erste Element, dass wir in die Queue einfügen, legen wir auf einen Stack (im Beispiel: „banana“). Um es aus der Queue zu entnehmen, holen wir es wieder vom Stack:

Mit dem zweiten Element funktioniert das so schon nicht mehr, denn der Stack arbeitet ja nach dem LIFO-Prinzip. Wenn z. B. „banana“ und „cherry“ auf dem Stack liegen, müssten wir „cherry“ als erstes wieder entnehmen:

Bei einer Queue wollen wir jedoch das zuerst eingefügte Element (also „banana“) auch als erstes wieder entnehmen.

Das ist mit einem Stack allein nicht möglich.

Stattdessen gehen wir beim Einfügen eines Elements in die Queue wie folgt vor:

1. Wir erzeugen einen temporären Stack (in der Grafik unten orange dargestellt) und verschieben alle Elemente des ursprünglichen Stacks auf den temporären Stack.
2. Wir legen das neue Element auf den ursprünglichen Stack.
3. Wir schieben alle Elemente zurück vom temporären Stack auf den ursprünglichen Stack. Der temporäre Stack wird dann nicht mehr benötigt.

Die folgende Grafik zeigt die drei Schritte:

Danach liegen die Elemente so auf dem Stack, dass wir als erstes das zuerst eingefügte Element „banana“ entnehmen können und dann das als zweites eingefügte Element „cherry“.

Das funktioniert so nicht nur mit zwei Elementen, sondern mit beliebig vielen. Die folgende Grafik zeigt, wie wir ein drittes Element „grape“ in die Queue einfügen:

Danach können wir die Elemente in First-in-First-out-Reihenfolge aus der Queue entnehmen, also erst die zuerst eingefügte „banana“, dann die „cherry“, und dann die zuletzt eingefügte „grape“.

Quellcode für die Queue mit Stacks

Der Quellcode für diesen Algorithmus benötigt nur wenige Zeilen Java-Code.

Als Stack verwende ich den im Stack-Tutorial vorgestellten ArrayStack. Genauso gut könntest du die JDK-Klasse Stack oder eine beliebige Deque-Implementierung, z. B. ein ArrayDeque, verwenden.

Du findest den Code in der Klasse StackQueue im GitHub-Repository.

public class StackQueue<E> implements Queue<E> {

  private final Stack<E> stack = new ArrayStack<>();

  @Override
  public void enqueue(E element) {
    // 1. Move elements from main stack to a temporary stack
    Stack<E> temporaryStack = new ArrayStack<>();
    while (!stack.isEmpty()) {
      temporaryStack.push(stack.pop());
    }

    // 2. Push new element on the main stack
    stack.push(element);

    // 3. Move elements back from temporary stack to main stack
    while (!temporaryStack.isEmpty()) {
      stack.push(temporaryStack.pop());
    }
  }

  @Override
  public E dequeue() {
    return stack.pop();
  }

  @Override
  public E peek() {
    return stack.peek();
  }

  @Override
  public boolean isEmpty() {
    return stack.isEmpty();
  }
}Code-Sprache: Java (java)

Beachte, dass hier nicht das java.util.Queue-Interface implementiert wird. Dieses erbt von java.util.Collection, sodass wir wesentlich mehr Methoden implementieren müssten.

Stattdessen habe ich für dieses Tutorial ein eigenes Queue-Interface geschrieben, das nur die Methoden enqueue(), dequeue(), peek() und isEmpty() definiert:

public interface Queue<E> {
  void enqueue(E element);
  E dequeue();
  E peek();
  boolean isEmpty();
}Code-Sprache: Java (java)

Zum Einsatz der Queue kannst du dir das Demo-Programm QueueDemo ansehen.

Ausblick

Im nächsten Teil des Tutorials erfährst du, wie man eine Queue mit einer verketteten Liste implementiert.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

Dieser Artikel gibt einen Überblick über alle im JDK verfügbaren Queue-Implementierungen mitsamt ihrer Eigenschaften sowie eine Entscheidungshilfe, für welche Einsatzzwecke welche Implementierung am besten geeignet ist.

Der Klassenname in der folgenden Tabelle ist jeweils mit demjenigen Artikel der Tutorial-Serie verlinkt, in dem die jeweilige Queue-Implementierung mit all ihren Eigenschaften im Detail erklärt wird.

Eine Erklärung der Begriffe blocking, non-blocking, fairness policy, bounded und unbounded findest du im Artikel über das BlockingQueue-Interface.

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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.

² Fail-fast: Der Iterator wirft eine ConcurrentModificationException, wenn während der Iteration Elemente in die Queue eingefügt oder aus dieser entnommen werden.

Wann solltest du welche Queue-Implementierung einsetzen?

Anhand der Eigenschaften der Queue-Implementierungen, die in den jeweiligen Artikeln beschrieben und in der Tabelle oben zusammengefasst sind, kannst du die richtige Queue für jeden Einsatzzweck finden.

Für den tagtäglichen Gebrauch der allgemeinen Queue-Implementierungen mache ich folgende Empfehlungen:

• ArrayDeque für single-threaded Anwendungen.
• ConcurrentLinkedQueue als threadsichere, nicht blockierende und unbounded Queue.
• ArrayBlockingQueue als threadsichere, blockierende, bounded Queue, sofern du niedrige bis mittlere Contention zwischen Producer- und Consumer-Threads erwartest.
• LinkedBlockingQueue als threadsichere, blockierende, bounded Queue, wenn du hohe Contention zwischen Producer- und Consumer-Threads erwartest (am besten testen, welche Implementierung für deinen Use Case performanter ist).

Hier das ganze noch einmal als Entscheidungsbaum:

Optimierte MPMC-, MPSC-, SPMC- und SPSC-Queues

Alle vom JDK angebotenen threadsicheren Queue-Implementierungen können bedenkenlos in multi-producer-multi-consumer-Umgebungen eingesetzt werden. Das bedeutet, dass ein oder mehrere schreibende Threads sowie ein oder mehrere lesende Threads nebenläufig auf die JDK-Queues zugreifen können.

Mit speziellen Mechanismen ist es möglich Queues so zu optimieren, dass der Overhead für das Gewährleisten der Threadsicherheit minimiert wird, wenn es eine Beschränkung auf einen lesenden und/oder einen schreibenden Thread gibt.

Dementsprechend unterscheidet man folgende vier Fälle:

• Multi-producer-multi-consumer (MPMC)
• Multi-producer-single-consumer (MPSC)
• Single-producer-multi-consumer (SPMC)
• Single-producer-single-consumer (SPSC)

Die Open-Source-Bibliothek JCTools bietet für alle vier Fälle hoch-optimierte Queue-Implementierungen.

Zusammenfassung und Ausblick

Dieser Artikel hat einen Überblick über alle in Java verfügbaren Queue-Implementierungen gegeben sowie eine Entscheidungshilfe, in welchen Fällen welche Queue einzusetzen ist.

In den nächsten Teilen dieser Serie zeige ich dir, wie du Queues selbst implementieren kannst, beginnend mit der Implementierung mit einem Stack.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

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 Klassendiagram 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 übergebe Element entgegennimmt:

• transfer(E e) – übermittelt das Element an einen Thread, der mit take() oder poll() auf ein Element wartet. Wenn solch ein Thread nicht existiert, blockiert die Methode solange, bis ein anderer Thread take() oder poll() aufruft.
• tryTransfer(E e) – übermittelt das Element an einen Thread, der mit take() oder 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 mit take() oder poll() auf ein Element wartet. Wenn solch ein Thread nicht existiert und auch nicht innerhalb der Wartezeit erscheint, gibt die Methode false zurück.

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()– bzw. tryTransfer()-Methoden 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:

¹ 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 Interators 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

Die LinkedTransferQueue wird im JDK nicht verwendet. Ursprünglich wurde sie für das im JDK 7 eingeführte Fork/Join-Framework implementiert, dann aber noch nicht dafür genutzt. Die Wahrscheinlichkeit von Bugs ist daher recht hoch, so dass du auf den Einsatz dieser Klasse verzichten solltest.

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 solange Elemente aus der Queue entnommen, bis diese wieder leer ist.

public class LinkedTransferQueueExample {
  public static void main(String[] args) 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));
  }
}Code-Sprache: Java (java)

Im folgenden siehst du die Ausgabe des Programms:

[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 = []Code-Sprache: Klartext (plaintext)

Man sieht sehr schön, wie zu Beginn zwei mal transfer() aufgerufen wird (aber nicht zurückkehrt), wie dann einmal put() aufgerufen wird (und zurückkehrt) und wie noch zwei mal transfer() aufgerufen wird (und nicht zurückkehrt).

Danach sehen wir, wie das erste Element entnommen wird und daraufhin auch transfer(1) zurückkehrt.

Dann wird das zweite Element entnommen und transfer(2) kehrt zurück.

Die Entnahme der 3 führt zu keiner weiteren Aktion, da diese mit put() in die Queue geschrieben wurde.

Nach der Entnahme der 4 und der 5 sieht man wieder schön, wie der jeweils zugehörige transfer()-Aufruf zurückkehrt.

Zusammenfassung und Ausblick

In diesem Artikel hast du das TransferQueue-Interface und die LinkedTransferQueue-Implementierung kennengelernt und an einem Beispiel gesehen, wie man diese einsetzen kann.

Im nächsten Teil dieser Tutorialserie findest du eine Zusammenfassung aller Queue-Implementierungen des JDK und eine Übersicht, in welchen Fällen du welche Implementierung verwenden solltest.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Artikel geht es um eine sehr spezielle Queue – die SynchronousQueue, deren Eigenschaften und Einsatzgebiete. Anhand eines Beispiels siehst du, wie man die SynchronousQueue einsetzt.

Hier befinden wir uns in der Klassenhierarchie:

SynchronousQueue Eigenschaften

Das “Synchronous” in der Klasse java.util.concurrent.SynchronousQueue ist nicht mit „synchronized“ zu verwechseln. Es bedeutet vielmehr, dass jede Enqueue-Operation auf eine korrespondierende Dequeue-Operation warten muss und jede Dequeue-Operation auf eine Enqueue-Operation.

Eine SynchronousQueue enthält niemals Elemente, auch dann nicht, wenn Enqueue-Operationen gerade auf Dequeue-Operationen warten. Analog dazu ist die Größe einer SynchronousQueue immer 0, und peek() liefert immer null zurück.

Die SynchronousQueue und die ArrayBlockingQueue sind die einzigen Queue-Implementierungen, die eine Fairness Policy anbieten. Hierbei gibt es eine Besonderheit: Wenn die Fairness Policy nicht aktiviert ist, werden blockierende Aufrufe laut Dokumentation in unspezifizierter Reihenfolge bedient. Tatsächlich ist es jedoch so, dass diese exakt in umgekehrter Reihenfolge bedient werden (also in LIFO-Reihenfolge), da intern ein Stack verwendet wird.

Die Eigenschaften der SynchronousQueue im Detail:

Einsatzempfehlung

Genau wie die DelayQueue und die LinkedTransferQueue habe ich auch die SynchronousQueue in eigenen Projekten noch nie direkt eingesetzt.

Sollten ihre Eigenschaften zu deinen Anforderungen passen, kannst du sie bedenkenlos verwenden. Im JDK wird die SynchronousQueue von Executors.newCachedThreadPool() als „work queue“ für den Executor eingesetzt; die Wahrscheinlichkeit von Bugs ist also äußerst gering.

SynchronousQueue Beispiel

Im folgende Beispiel (→ Code auf GitHub) werden zunächst drei Threads gestartet, die SynchronousQueue.put() aufrufen, danach sechs Threads, die SynchronousQueue.take() aufrufen und anschließend noch einmal drei Threads, die SynchronousQueue.put() ausführen:

public class SynchronousQueueExample {
  private static final boolean FAIR = false;

  public static void main(String[] args) throws InterruptedException {
    BlockingQueue<Integer> queue = new SynchronousQueue<>(FAIR);

    // Start 3 producing threads
    for (int i = 0; i < 3; i++) {
      int element = i; // Assign to an effectively final variable
      new Thread(() -> enqueue(queue, element)).start();
      Thread.sleep(250);
    }

    // Start 6 consuming threads
    for (int i = 0; i < 6; i++) {
      new Thread(() -> dequeue(queue)).start();
      Thread.sleep(250);
    }

    // Start 3 more producing threads
    for (int i = 3; i < 6; i++) {
      int element = i; // Assign to an effectively final variable
      new Thread(() -> enqueue(queue, element)).start();
      Thread.sleep(250);
    }
  }

  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,
        "[%-9s] %s%n",
        Thread.currentThread().getName(),
        String.format(format, args));
  }
}
Code-Sprache: Java (java)

In der Ausgabe kannst du sehen, wie die ersten drei Aufrufe von put() durch die Threads 0, 1 und 2 solange blockieren, bis die eingefügten Elemente mit take() durch die Threads 3, 4 und 5 in umgekehrter Reihenfolge entnommen werden.

Danach blockieren die drei folgenden Aufrufe von take() (Threads 6, 7, 8) solange bis mit put() drei weitere Elemente in die Queue geschrieben wurden (Threads 9, 10, 11).

Die Queue bleibt während der gesamten Zeit leer.

[Thread-0 ] Calling queue.put(0) (queue = [])...
[Thread-1 ] Calling queue.put(1) (queue = [])...
[Thread-2 ] Calling queue.put(2) (queue = [])...
[Thread-3 ]     Calling queue.take() (queue = [])...
[Thread-3 ]     queue.take() returned 2 (queue = [])
[Thread-2 ] queue.put(2) returned (queue = [])
[Thread-4 ]     Calling queue.take() (queue = [])...
[Thread-4 ]     queue.take() returned 1 (queue = [])
[Thread-1 ] queue.put(1) returned (queue = [])
[Thread-5 ]     Calling queue.take() (queue = [])...
[Thread-5 ]     queue.take() returned 0 (queue = [])
[Thread-0 ] queue.put(0) returned (queue = [])
[Thread-6 ]     Calling queue.take() (queue = [])...
[Thread-7 ]     Calling queue.take() (queue = [])...
[Thread-8 ]     Calling queue.take() (queue = [])...
[Thread-9 ] Calling queue.put(3) (queue = [])...
[Thread-9 ] queue.put(3) returned (queue = [])
[Thread-8 ]     queue.take() returned 3 (queue = [])
[Thread-10] Calling queue.put(4) (queue = [])...
[Thread-10] queue.put(4) returned (queue = [])
[Thread-7 ]     queue.take() returned 4 (queue = [])
[Thread-11] Calling queue.put(5) (queue = [])...
[Thread-11] queue.put(5) returned (queue = [])
[Thread-6 ]     queue.take() returned 5 (queue = [])Code-Sprache: Klartext (plaintext)

Wenn du die Konstante FAIR auf true setzt, wirst du sehen, wie die Elemente nicht in LIFO-, sondern in FIFO-Reihenfolge entnommen werden.

Zusammenfassung und Ausblick

In diesem Artikel hast du die SynchronousQueue kennengelernt – eine Queue, die niemals Elemente enthält, sondern diese direkt von den enqueuenden Threads an die dequeuenden Threads übergibt.

Im nächsten Teil geht es um die letzte Queue-Impementierung dieser Tutorial-Serie: die LinkedTransferQueue.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem und den nächsten Teilen dieser Tutorialserie geht es um Queues für spezielle Einsatzzwecke. Wir beginnen mit der DelayQueue, einer Queue, die die Elemente nach Ablaufzeit sortiert herausgibt.

Hier befinden wir uns in der Klassenhierarchie:

DelayQueue Eigenschaften

Die Klasse java.util.concurrent.DelayQueue ist – genau wie die PriorityQueue, die sie intern verwendet – keine FIFO-Queue. Es wird nicht das Element entnommen, das sich am längsten in der Queue befindet. Stattdessen wird ein Element dann entnommen, wenn eine diesem Element zugeordnete Wartezeit („delay“) abgelaufen ist.

Dazu müssen die Elemente das Interface java.util.concurrent.Delayed und dessen getDelay​()-Methode implementieren. Diese Methode liefert bei jeden Aufruf die restliche Wartezeit zurück, die noch ablaufen muss, bis das Element aus der Queue entnommen werden darf.

DelayQueue ist blockierend, aber unbounded, kann also beliebig viele Elemente aufnehmen und blockiert nur bei der Entnahme (bis die Wartezeit abgelaufen ist), nicht beim Einfügen.

Threadsicherheit wird durch pessimistisches Locking über ein einzelnes ReentrantLock sichergestellt.

Die Eigenschaften im Detail:

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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

Ich habe die DelayQueue noch nie benötigt und kann sie für keinen mir bekannten, sinnvollen Einsatzzweck empfehlen. Sie wird im JDK nur ein einziges Mal genutzt (in einer alten Swing-Klasse, die mit einem ScheduledExecutorService wahrscheinlich eleganter hätte implementiert werden können). Von daher ist nicht ausgeschlossen, dass sie unentdeckte Fehler enthält.

DelayQueue Beispiel

In folgendem Beispiel (→ Code auf GitHub) wird eine DelayQueue mit Instanzen der Klasse DelayedElement gefüllt, welche eine Zufallszahl enthalten sowie ein zufälliges initiales Delay zwischen 100 und 1.000 ms. Danach wird so oft poll() aufgerufen, bis die Queue wieder leer ist.

public class DelayQueueExample {
  public static void main(String[] args) {
    BlockingQueue<DelayedElement<Integer>> queue = new DelayQueue<>();
    ThreadLocalRandom random = ThreadLocalRandom.current();
    long startTime = System.currentTimeMillis();

    // Enqueue random numbers with random initial delays
    for (int i = 0; i < 7; i++) {
      int randomNumber = random.nextInt(10, 100);
      int initialDelayMillis = random.nextInt(100, 1000);
      DelayedElement<Integer> element = 
          new DelayedElement<>(randomNumber, initialDelayMillis);
      queue.offer(element);
      System.out.printf(
          "[%3dms] queue.offer(%s)   --> queue = %s%n",
          System.currentTimeMillis() - startTime, element, queue);
    }

    // Dequeue all elements
    while (!queue.isEmpty()) {
      try {
        DelayedElement<Integer> element = queue.take();
        System.out.printf(
            "[%3dms] queue.poll() = %s --> queue = %s%n",
            System.currentTimeMillis() - startTime, element, queue);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }
}Code-Sprache: Java (java)

Und hier die zugehörige Klasse DelayedElement (→ Code auf GitHub). Die Sortierung ist – um den Code nicht noch länger zu machen – nicht stabil. D. h. wenn zwei Elemente mit derselben Wartezeit in die Queue gestellt werden, werden sie in zufälliger Reihenfolge wieder entnommen.

public class DelayedElement<E extends Comparable<E>> implements Delayed {
  private final E e;
  private final long initialDelayMillis;
  private final long expiration;

  public DelayedElement(E e, long initialDelayMillis) {
    this.e = e;
    this.initialDelayMillis = initialDelayMillis;
    this.expiration = System.currentTimeMillis() + initialDelayMillis;
  }

  @Override
  public long getDelay(TimeUnit unit) {
    long remainingDelayMillis = expiration - System.currentTimeMillis();
    return unit.convert(remainingDelayMillis, TimeUnit.MILLISECONDS);
  }

  @Override
  public int compareTo(Delayed o) {
    DelayedElement<?> other = (DelayedElement<?>) o;
    return Long.compare(expiration, other.expiration);
  }

  @Override
  public String toString() {
    return "{%s, %dms}".formatted(e, initialDelayMillis);
  }
}
Code-Sprache: Java (java)

Hier eine beispielhafte Ausgabe des Programms. Es ist gut zu sehen, wie die Queue zwar nicht sortiert ist¹, das Element mit der kürzesten Wartezeit sich jedoch immer vorne (links) befindet und dass die Elemente (ungefähr) nach Ablauf ihrer jeweiligen Wartezeit entnommen werden:

[  1ms] queue.offer({19, 713ms})   --> queue = [{19, 713ms}]
[ 28ms] queue.offer({15, 643ms})   --> queue = [{15, 643ms}, {19, 713ms}]
[ 29ms] queue.offer({35, 253ms})   --> queue = [{35, 253ms}, {19, 713ms}, {15, 643ms}]
[ 29ms] queue.offer({11, 781ms})   --> queue = [{35, 253ms}, {19, 713ms}, {15, 643ms}, {11, 781ms}]
[ 29ms] queue.offer({17, 844ms})   --> queue = [{35, 253ms}, {19, 713ms}, {15, 643ms}, {11, 781ms}, {17, 844ms}]
[ 29ms] queue.offer({40, 490ms})   --> queue = [{35, 253ms}, {19, 713ms}, {40, 490ms}, {11, 781ms}, {17, 844ms}, {15, 643ms}]
[ 30ms] queue.offer({39, 119ms})   --> queue = [{39, 119ms}, {19, 713ms}, {35, 253ms}, {11, 781ms}, {17, 844ms}, {15, 643ms}, {40, 490ms}]
[151ms] queue.poll() = {39, 119ms} --> queue = [{35, 253ms}, {19, 713ms}, {40, 490ms}, {11, 781ms}, {17, 844ms}, {15, 643ms}]
[283ms] queue.poll() = {35, 253ms} --> queue = [{40, 490ms}, {19, 713ms}, {15, 643ms}, {11, 781ms}, {17, 844ms}]
[520ms] queue.poll() = {40, 490ms} --> queue = [{15, 643ms}, {19, 713ms}, {17, 844ms}, {11, 781ms}]
[673ms] queue.poll() = {15, 643ms} --> queue = [{19, 713ms}, {11, 781ms}, {17, 844ms}]
[716ms] queue.poll() = {19, 713ms} --> queue = [{11, 781ms}, {17, 844ms}]
[811ms] queue.poll() = {11, 781ms} --> queue = [{17, 844ms}]
[874ms] queue.poll() = {17, 844ms} --> queue = []Code-Sprache: Klartext (plaintext)

¹ Tatsächlich sieht man die Reihenfolge der Elemente in der Array-Repräsentation des Min-Heaps.

Zusammenfassung und Ausblick

In diesem Artikel hast du alles über die DelayQueue, ihre Eigenschaften und ihre Verwendung erfahren.

Im nächsten Teil dieser Serie stelle ich dir weitere Spezialqueue vor – eine die niemals Elemente enthält: die SynchronousQueue.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Artikel erfährst du wie die PriorityBlockingQueue funktioniert und welche Eigenschaften sie hat. Anhand eines Beispiels siehst du, wie man sie einsetzt.

Hier befinden wir uns in der Klassenhierarchie:

PriorityBlockingQueue Eigenschaften

Bei der java.util.concurrent.PriorityBlockingQueue handelt es sich um eine threadsichere und blockierende Variante der PriorityQueue. In dem verlinkten Artikel erfährst du auch, was eine Priotity Queue ist.

Wie bei der PriorityQueue werden die Elemente in einem Array gespeichert, das einen Min-Heap repräsentiert; der Iterator durchläuft die Elemente in entsprechender Reihenfolge.

Threadsicherheit wird durch ein einzelnes ReentrantLock sichergestellt.

Die PriorityBlockingQueue ist nicht bounded, sie hat also keine Kapazitätsgrenze. Das bedeutet, dass put(e) und offer(e, time, unit) niemals blockieren. Nur die Dequeue-Operationen take() und poll(time, unit) blockieren, wenn die Queue leer ist.

Die Eigenschaften im Detail:

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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

Die PriorityBlockingQueue wird im JDK nicht genutzt. Es ist daher nicht auszuschließen, dass sie Bugs enthält. Wenn du eine Queue mit entsprechenden Eigenschaften benötigst und die PriorityBlockingQueue verwendest, solltest du deine Anwendung intensiv testen.

PriorityBlockingQueue Beispiel

Das folgenden Beispiel zeigt, wie eine PriorityBlockingQueue angelegt wird und wie mehrere Threads lesend und schreibend darauf zugreifen (→ Code auf GitHub).

Lesende Threads starten alle 3 Sekunden, beginnend sofort nach dem Erstellen der Queue.

Schreibende Threads starten nach 3,5 Sekunden (so dass bereits zwei lesende Threads warten) und schreiben sekündlich einen Zufallswert in die Queue.

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

  public static void main(String[] args) throws InterruptedException {
    BlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
    ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);

    // Start reading from the queue immediately, every 3 seconds
    for (int i = 0; i < 8; 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 some more elements and their order in the queue)
    for (int i = 0; i < 8; i++) {
      int delayMillis = 3500 + i * 1000;
      pool.schedule(() -> enqueue(queue), delayMillis, TimeUnit.MILLISECONDS);
    }

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

  private static void enqueue(BlockingQueue<Integer> queue) {
    int element = ThreadLocalRandom.current().nextInt(10, 100);
    log("Calling queue.put(%d) (queue = %s)...", element, queue);
    try {
      queue.put(element);
      log("queue.put(%d) returned (queue = %s)", element, queue);
    } catch (InterruptedException ex) {
      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));
  }
}Code-Sprache: Java (java)

Im folgenden siehst du eine beispielhafte Ausgabe des Programms:

[ 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-6 ] Calling queue.put(87) (queue = [])...
[ 3.5s] [pool-1-thread-6 ] queue.put(87) returned (queue = [])
[ 3.5s] [pool-1-thread-1 ]     queue.take() returned 87 (queue = [])
[ 4.5s] [pool-1-thread-9 ] Calling queue.put(89) (queue = [])...
[ 4.5s] [pool-1-thread-9 ] queue.put(89) returned (queue = [])
[ 4.5s] [pool-1-thread-2 ]     queue.take() returned 89 (queue = [])
[ 5.5s] [pool-1-thread-7 ] Calling queue.put(31) (queue = [])...
[ 5.5s] [pool-1-thread-7 ] queue.put(31) returned (queue = [31])
[ 6.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [31])...
[ 6.0s] [pool-1-thread-4 ]     queue.take() returned 31 (queue = [])
[ 6.5s] [pool-1-thread-5 ] Calling queue.put(71) (queue = [])...
[ 6.5s] [pool-1-thread-5 ] queue.put(71) returned (queue = [71])
[ 7.5s] [pool-1-thread-8 ] Calling queue.put(15) (queue = [71])...
[ 7.5s] [pool-1-thread-8 ] queue.put(15) returned (queue = [15, 71])
[ 8.5s] [pool-1-thread-10] Calling queue.put(33) (queue = [15, 71])...
[ 8.5s] [pool-1-thread-10] queue.put(33) returned (queue = [15, 71, 33])
[ 9.0s] [pool-1-thread-3 ]     Calling queue.take() (queue = [15, 71, 33])...
[ 9.0s] [pool-1-thread-3 ]     queue.take() returned 15 (queue = [33, 71])
[ 9.5s] [pool-1-thread-6 ] Calling queue.put(58) (queue = [33, 71])...
[ 9.5s] [pool-1-thread-6 ] queue.put(58) returned (queue = [33, 71, 58])
[10.5s] [pool-1-thread-1 ] Calling queue.put(19) (queue = [33, 71, 58])...
[10.5s] [pool-1-thread-1 ] queue.put(19) returned (queue = [19, 33, 58, 71])
[12.0s] [pool-1-thread-9 ]     Calling queue.take() (queue = [19, 33, 58, 71])...
[12.0s] [pool-1-thread-9 ]     queue.take() returned 19 (queue = [33, 71, 58])
[15.0s] [pool-1-thread-2 ]     Calling queue.take() (queue = [33, 71, 58])...
[15.0s] [pool-1-thread-2 ]     queue.take() returned 33 (queue = [58, 71])
[18.0s] [pool-1-thread-7 ]     Calling queue.take() (queue = [58, 71])...
[18.0s] [pool-1-thread-7 ]     queue.take() returned 58 (queue = [71])
[21.0s] [pool-1-thread-4 ]     Calling queue.take() (queue = [71])...
[21.0s] [pool-1-thread-4 ]     queue.take() returned 71 (queue = [])Code-Sprache: Klartext (plaintext)

Was kann man in dieser Beispielausgabe erkennen?

Zunächst einmal siehst du, wie nach 0,0 s und 3,0 s die Threads 1 und 2 beim Aufruf von take() blockieren, da die Queue leer ist.

Nach 3,5 s schreibt Thread 6 die 87 in die Queue. Sofort im Anschluss wacht der zuvor blockierte Thread 1 auf und entnimmt die 87 wieder.

Nach 4,5 s schreibt Thread 9 die 89 in die Queue, die sofort von Thread 2 wieder entnommen wird.

Nach 5,5 s wird die 31 in die Queue geschrieben, die nach 6,0 s wieder entnommen wird.

Nach 6,5 s, 7,5 s und 8,5 s werden die 71, die 15 und die 33 in die Queue geschrieben. Du siehst, wie jeweils das kleinste Element vorne (links) in der Queue steht.

Nach 9,0 s wird das kleinste Elemente, die 15, entnommen. Daraufhin steht das nächstkleinere Element, die 33 am Kopf der Queue.

Nach 9,5 s und 10,5 s werden zwei weitere Elemente, 58 und 19, in die Queue geschrieben. Du siehst wieder gut, wie jeweils das kleinste Element am Kopf der Queue steht.

Die Queue enthält nun vier Elemente. Es werden keine weiteren Elemente in die Queue geschrieben und die existierenden Elemente entsprechend ihrer Priorität entnommen.

Zusammenfassung und Ausblick

In diesem Artikel hast du erfahren, welche Eigenschaften die PriorityBlockingQueue hat und wie diese eingesetzt wird.

Ab dem nächsten Teil der Tutorial-Serie stelle ich dir einige Queue-Implementierungen für Sonderfälle vor, beginnend mit der DelayQueue.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

Im 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 maximale Kapazität), entsprechend blockierend und bietet eine Fairness Policy (d. h. blockierende Methoden werden in der Reihenfolge bedient, in der sie aufgerufen wurden).

Die Eigenschaften in der Übersicht:

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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

Aufgrund der möglicherweise hohen Contention bei gleichzeitigem Schreib- und Lesezugriff solltest du – wenn du eine blockierende, threadsichere Queue benötigst – für deinen speziellen Einsatzweck testen, ob evtl. eine LinkedBlockingQueue performanter ist. Diese basiert zwar auf einer verketteten Liste, verwendet allerdings zum Schreiben und Lesen zwei separate ReentrantLocks, was die Zugangskonflikte reduziert.

ArrayBlockingQueue Beispiel

Im folgenden Beispiel erzeugen wir eine ArrayBlockingQueue mit der Kapazität 3. 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();

  public static void main(String[] args) throws InterruptedException {
    BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(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));
  }
}Code-Sprache: Java (java)

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 = [])Code-Sprache: Klartext (plaintext)

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.

Ist ArrayBlockingQueue threadsicher?

Ja, ArrayBlockingQueue ist threadsicher.

Die Threadsicherheit von ArrayBlockingQueue wird durch ein einzelnes ReentrantLock gewährleistet. Dieses wird sowohl für den Kopf als auch für das Ende der Queue verwendet, so dass es bei gleichzeitigen Schreib- und Lesezugriffen zu Zugriffskonflikten („thread contention“) zwischen Producer- und Consumer-Threads kommen kann.

Explizite Locks wie ReentrantLock sind hauptsächlich für Einsatzgebiete geeignet, in denen es zu hoher Thread Contention kommt. Bei niedriger bis moderater Contention ist optimistisches Locking performanter.

Unterschiede zu anderen Queues:

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

Zusammenfassung und Ausblick

Dieser Artikel hat dir die ArrayBlockingQueue vorgestellt. Diese Queue ist threadsicher, blockierend und bounded. Anhand eines Beispiels hast du gesehen, wie du ArrayBlockingQueue einsetzen kannst.

Wie der Name schon sagt, basiert diese Queue auf einem Array. Das auf einer verketteten Liste basierende Pendant – 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.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Teil der Tutorialserie geht es um die LinkedBlockingQueue. Du wirst deren speziellen 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.), bounded und blockierend.

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.)

Die Queue-Eigenschaften im Detail:

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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

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

Die Klasse LinkedBlockingQueue wird übrigens von Executors.newFixedThreadPool() und Executors.newSingleThreadedExecutor() als „work queue“ für den Executor verwendet. Sie wird somit intensiv genutzt, was die Wahrscheinlichkeit für Bugs äußerst gering hält.

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();

  public static void main(String[] args) 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));
  }
}Code-Sprache: Java (java)

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 = [])Code-Sprache: Klartext (plaintext)

Da wir erst mit dem Schreiben beginnen, nachdem bereits zwei Threads take() aufrufen, 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 ReentrantLock für Schreib- und Leseoperationen gewährleistet. So kann es zu keiner Contention (Zugriffskonflikten) zwischen Producer- und Consumer-Threads kommen.

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, so dass dort Zugriffskonflikte zwischen Producer- und Consumer-Threads möglich sind.

LinkedBlockingQueue Zeitkomplexität

Wie bei allen Queues 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 intern ein AtomicInteger, das beim Einfügen und Entnehmen aktualisiert wird, und somit die Größe mit konstantem Aufwand verfügbar hält.

Zusammenfassung und Ausblick

In diesem Artikel hast du die LinkedBlockingQueue kennengelernt – eine threadsichere, blockierende, bounded Queue. 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.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Teil der Tutorialserie stelle ich dir eine Queue vor, die streng genommen gar keine Queue ist: die PriorityQueue.

Wir befinden uns hier in der Klassenhierarchie:

Was ist eine Priority Queue?

Eine Priority Queue (deutsch: Vorrangwarteschlange) ist keine Queue im klassischen Sinne. Denn die Elemente werden nicht in FIFO-Reihenfolge entnommen, sondern entsprechend ihrer Priorität. Es wird immer das Element mit der höchsten Priorität zuerst entnommen – unabhängig davon, wann es in die Queue eingefügt wurde.

Das folgende Beispiel zeigt eine Priority Queue mit Elementen der Priorität 10 (höchste Priorität), 20, usw. bis 80 (niedrigste Priorität). Ein weiteres Element mit Priorität 45 wird in die Queue eingefügt. Die Queue sorgt dann automatisch dafür, dass dieses Element nach dem Element mit der Priorität 40 und vor dem Element mit der Priorität 50 entnommen wird.

Mit welcher Datenstruktur wird eine Priority Queue implementiert?

Priority Queues werden in der Regel mit einem Heap implementiert.

Im letzten Teil dieser Tutorialserie werde ich dir zeigen, wie du selbst eine Priority Queue mit einem Heap implementieren kannst.

Eigenschaften der Java-PriorityQueue

Bei der Klasse java.util.PriorityQueue ergibt sich die Dequeue-Reihenfolge entweder aus der natürlichen Ordnung¹ der Elemente oder entsprechend eines dem Constructor übergebenen Comparators¹. Die zugrundeliegende Datenstruktur ist ein Min-Heap, d. h. am Kopf der Queue befindet sich immer das kleinste Element.

Die Sortierung ist dabei nicht stabil, d. h. zwei Elemente, die entsprechend der Sortierreihenfolge an gleicher Position stehen, werden nicht zwangsläufig in derselben Reihenfolge entnommen, wie sie in die Queue eingefügt wurden.

PriorityQueue ist weder threadsicher noch blockierend. Als threadsicheres, blockierendes Pendant gibt es die PriorityBlockingQueue.

Die Queue-Eigenschaften sind:

Die PriorityQueue verstößt übrigens nicht gegen das Liskovsche Substitutionsprinzip (Liskov substitution principle, LSP). Denn die Dokumentation des Queue-Interfaces besagt: „Queues typically, but do not necessarily, order elements in a FIFO (first-in-first-out) manner.“

¹ Alles über die „natürliche Ordnung“ von Objekten und die Sortierung per Comparator liest du im Artikel „Vergleichen von Objekten in Java„.

² Fail-fast: Der Iterator wirft eine ConcurrentModificationException, wenn während der Iteration Elemente in die Queue eingefügt oder aus dieser entnommen werden.

Einsatzempfehlung

Die PriorityQueue kann genau dann eingesetzt werden, wenn eine nicht threadsichere Queue mit oben beschriebener Dequeue-Reihenfolge benötigt wird.

Allerdings ist zu beachten, dass die PriorityQueue im JDK nur an sehr wenigen Stellen eingesetzt wird und somit eine gewisse Wahrscheinlichkeit für das Vorhandensein von Bugs existiert (was wenig genutzt wird, wird auch wenig getestet).

PriorityQueue Beispiel

Das folgende Beispiel zeigt, wie in Java eine Priority Queue erstellt wird und wie mehrere Zufallszahlen in die Queue geschrieben und dann wieder entnommen werden (→ Code auf GitHub).

Wir geben keinen Comparator an, d. h. die Integer-Elemente werden entsprechend ihrer natürlichen Ordnung sortiert.

public class PriorityQueueExample {
  public static void main(String[] args) {
    Queue<Integer> queue = new PriorityQueue<>();

    // Enqueue random numbers
    for (int i = 0; i < 8; i++) {
      int element = ThreadLocalRandom.current().nextInt(100);
      queue.offer(element);
      System.out.printf("queue.offer(%2d)    -->  queue = %s%n", element, queue);
    }

    // Dequeue all elements
    while (!queue.isEmpty()) {
      Integer element = queue.poll();
      System.out.printf("queue.poll() = %2d  -->  queue = %s%n", element, queue);
    }
  }
}Code-Sprache: Java (java)

Es folgt eine beispielhafte Ausgabe des Programms.

queue.offer(80)    -->  queue = [80]
queue.offer(14)    -->  queue = [14, 80]
queue.offer(10)    -->  queue = [10, 80, 14]
queue.offer(50)    -->  queue = [10, 50, 14, 80]
queue.offer( 9)    -->  queue = [9, 10, 14, 80, 50]
queue.offer(58)    -->  queue = [9, 10, 14, 80, 50, 58]
queue.offer(41)    -->  queue = [9, 10, 14, 80, 50, 58, 41]
queue.offer( 1)    -->  queue = [1, 9, 14, 10, 50, 58, 41, 80]
queue.poll() =  1  -->  queue = [9, 10, 14, 80, 50, 58, 41]
queue.poll() =  9  -->  queue = [10, 41, 14, 80, 50, 58]
queue.poll() = 10  -->  queue = [14, 41, 58, 80, 50]
queue.poll() = 14  -->  queue = [41, 50, 58, 80]
queue.poll() = 41  -->  queue = [50, 80, 58]
queue.poll() = 50  -->  queue = [58, 80]
queue.poll() = 58  -->  queue = [80]
queue.poll() = 80  -->  queue = []Code-Sprache: Klartext (plaintext)

Es ist gut zu sehen:

• wie acht Elemente in die Priority Queue eingefügt werden,
• wie die Elemente in der Priority Queue in vermeintlich zufälliger Reihenfolge dargestellt werden (tatsächlich handelt es sich um die Array-Repräsentation des Min-Heaps),
• dass das kleinste Element immer am Kopf der Queue (links) ist,
• wie die Elemente in aufsteigender Reihenfolge entnommen werden.

PriorityQueue mit einem Comparator

Im vorherigen Beispiel haben wir eine PriorityQueue über den Default-Konstruktor erstellt. Dies führt dazu, dass die Elemente entsprechend ihrer natürlichen Ordnung sortiert werden.

Wir können aber auch einen benutzerdefinierten Comparator für die Priority Queue angeben. Im folgenden Beispiel erstellen wir dazu Tasks mit je einem Namen und einer Priorität. Diese Tasks sollen nach Priorität sortiert wieder entnommen werden.

Den Comparator dafür definieren wir ganz einfach als:

Comparator<Task> comparator = Comparator.comparing(Task::name);Code-Sprache: Java (java)

Falls du mit dieser Schreibweise nicht vertraut bist – sie erstellt einen Comparator, der Tasks nach Priorität sortiert. Das ist doch viel besser lesbar als der folgende, mit einem Lambda, definierte Comparator:

Comparator<Task> comparator =
    (task1, task2) -> Integer.compare(task1.priority(), task2.priority());Code-Sprache: Java (java)

Hier der vollständige Beispiel-Code (Klasse PriorityQueueWithCustomComparatorExample auf GitHub):

public class PriorityQueueWithCustomComparatorExample {
  public static void main(String[] args) {
    Comparator<Task> comparator = Comparator.comparingInt(Task::priority);
    Queue<Task> queue = new PriorityQueue<>(comparator);

    // Enqueue tasks with random priorities
    for (int i = 0; i < 5; i++) {
      String name = "Task " + (i + 1);
      int priority = ThreadLocalRandom.current().nextInt(10, 100);
      Task task = new Task(name, priority);
      queue.offer(task);
      System.out.printf("queue.offer(%s)    -->  queue = %s%n", task, queue);
    }

    // Dequeue all elements
    while (!queue.isEmpty()) {
      System.out.printf("queue.poll() = %s  -->  queue = %s%n", queue.poll(), queue);
    }
  }

  private record Task(String name, int priority) {
    @Override
    public String toString() {
      return name + " (prio " + priority + ")";
    }
  }
}Code-Sprache: Java (java)

Eine beispielshafte Ausgabe sieht wie folgt aus:

queue.offer(Task 1 (prio 93))    -->  queue = [Task 1 (prio 93)]
queue.offer(Task 2 (prio 76))    -->  queue = [Task 2 (prio 76), Task 1 (prio 93)]
queue.offer(Task 3 (prio 92))    -->  queue = [Task 2 (prio 76), Task 1 (prio 93), Task 3 (prio 92)]
queue.offer(Task 4 (prio 51))    -->  queue = [Task 4 (prio 51), Task 2 (prio 76), Task 3 (prio 92), Task 1 (prio 93)]
queue.offer(Task 5 (prio 35))    -->  queue = [Task 5 (prio 35), Task 4 (prio 51), Task 3 (prio 92), Task 1 (prio 93), Task 2 (prio 76)]
queue.poll() = Task 5 (prio 35)  -->  queue = [Task 4 (prio 51), Task 2 (prio 76), Task 3 (prio 92), Task 1 (prio 93)]
queue.poll() = Task 4 (prio 51)  -->  queue = [Task 2 (prio 76), Task 1 (prio 93), Task 3 (prio 92)]
queue.poll() = Task 2 (prio 76)  -->  queue = [Task 3 (prio 92), Task 1 (prio 93)]
queue.poll() = Task 3 (prio 92)  -->  queue = [Task 1 (prio 93)]
queue.poll() = Task 1 (prio 93)  -->  queue = []
Code-Sprache: Klartext (plaintext)

Wir können gut erkennen, wie die Tasks entsprechend ihrer Priorität (Prio 51 zuerst, Prio 93 zuletzt) entnommen werden.

Um die Tasks nach Name alphabetisch sortiert zu entnehmen, bräuchten wir lediglich den Comparator ändern in:

Comparator<Task> comparator = Comparator.comparing(Task::name);Code-Sprache: Java (java)

PriorityQueue Zeitkomplexität

Der Aufwand für die Enqueue- und Dequeue-Operationen bei der Java-PriorityQueue gleicht dem Aufwand für das Einfügen in und Entnehmen aus einem Heap.

Die Zeitkomplexität für beide Operationen lautet somit: O(n log n)

Durch die Verwendung eines Heaps befindet sich das Element mit der höchsten Priority immer automatisch am Kopf der Queue und kann mit konstantem Aufwand entnommen werden.

Die Zeitkomplexität für die Peek-Operation beträgt somit: O(1)

(Hier findest du eine Einführung in das Thema Zeitkomplexität.)

Zusammenfassung und Ausblick

Dieser Artikel hat erklärt, was eine Priority Queue im allgemeinen ist, welche Eigenschaften die Java-PriorityQueue hat, wann man sie einsetzt, wie man die Dequeue-Reihenfolge mit einem benutzerdefinierten Comparator festlegt und welche Zeitkomplexitäten die Operationen der Priority Queue aufweisen.

Im folgenden Artikel kommen wir zur ersten blockierenden Queue, der LinkedBlockingQueue.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Artikel erfährst du alles über die ConcurrentLinkedQueue, deren Eigenschaften und Einsatzszenarien. Anhand eines Beispiels siehst du, wie man ConcurrentLinkedQueue einsetzt.

Hier befinden wir uns in der Klassenhierarchie:

ConcurrentLinkedQueue Eigenschaften

Die Klasse java.util.concurrent.ConcurrentLinkedQueue basiert auf einer einfach verketteten Liste und ist – wie die meisten Queue-Implementierungen – threadsicher (s. u.).

(Die einzige nicht threadsichere Queue ist PriorityQueue – sowie die Deques ArrayDeque und LinkedList, welche auch das Queue-Interface implementieren. Dazu mehr in der nächsten Tutorial-Serie „Deques“.)

Da die Länge einer verketteten Liste nur aufwändig zu bestimmen ist, ist ConcurrentLinkedQueue unbounded. ConcurrentLinkedQueue bietet außerdem keine blockierenden Operationen.

Die Eigenschaften im Detail:

¹ Weakly consistent: Alle Elemente, die zum Zeitpunkt der Erzeugung des Interators 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

ConcurrentLinkedQueue ist eine gute Wahl, wenn eine threadsichere, nicht blockierende und unbounded Queue benötigt wird.

Dies gilt trotz meiner allgemeinen Empfehlung Array-basierte Datenstrukturen solchen vorzuziehen, die mit verketteten Listen implementiert sind.

Array-basierte Alternativen sind:

• Das im Folgetutorial über Deques beschriebene ArrayDeque – dieses ist allerdings nicht thread-safe.
• Die später in diesem Tutorial beschriebe ArrayBlockingQueue – diese ist zum einen bounded, und zum anderen wird die Threadsicherheit über ein einzelnes ReentrantLock implementiert. Dies ist für die meisten Einsatzszenarien (mit niedriger bis mittlerer Contention) unperformanter ist als optimistisches Locking.

ConcurrentLinkedQueue Beispiel

Das folgende Beispiel demonstriert die Threadsicherheit von ConcurrentLinkedDeque. Vier schreibende und drei lesende Threads fügen nebenläufig Elemente in die Queue ein und entnehmen sie wieder (→ Code auf GitHub):

public class ConcurrentLinkedQueueExample {

  private static final int NUMBER_OF_PRODUCERS = 4;
  private static final int NUMBER_OF_CONSUMERS = 3;
  private static final int NUMBER_OF_ELEMENTS_TO_PUT_INTO_QUEUE_PER_THREAD = 5;
  private static final int MIN_SLEEP_TIME_MILLIS = 500;
  private static final int MAX_SLEEP_TIME_MILLIS = 2000;

  private static final int POISON_PILL = -1;

  public static void main(String[] args) throws InterruptedException {
    Queue<Integer> queue = new ConcurrentLinkedQueue<>();

    // Start producers
    CountDownLatch producerFinishLatch = new CountDownLatch(NUMBER_OF_PRODUCERS);
    for (int i = 0; i < NUMBER_OF_PRODUCERS; i++) {
      createProducerThread(queue, producerFinishLatch).start();
    }

    // Start consumers
    for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
      createConsumerThread(queue).start();
    }

    // Wait until all producers are finished
    producerFinishLatch.await();

    // Put poison pills on the queue (one for each consumer)
    for (int i = 0; i < NUMBER_OF_CONSUMERS; i++) {
      queue.offer(POISON_PILL);
    }

    // We'll let the program end when all consumers are finished
  }

  private static Thread createProducerThread(
      Queue<Integer> queue, CountDownLatch finishLatch) {
    return new Thread(
        () -> {
          ThreadLocalRandom random = ThreadLocalRandom.current();
          for (int i = 0; i < NUMBER_OF_ELEMENTS_TO_PUT_INTO_QUEUE_PER_THREAD; i++) {
            sleepRandomTime();

            Integer element = random.nextInt(1000);
            queue.offer(element);
            System.out.printf(
                "[%s] queue.offer(%3d)        --> queue = %s%n",
                Thread.currentThread().getName(), element, queue);
          }

          finishLatch.countDown();
        });
  }

  private static Thread createConsumerThread(Queue<Integer> queue) {
    return new Thread(
        () -> {
          while (true) {
            sleepRandomTime();

            Integer element = queue.poll();
            System.out.printf(
                "[%s]     queue.poll() = %4d --> queue = %s%n",
                Thread.currentThread().getName(), element, queue);

            // End the thread when a poison pill is detected
            if (element != null && element == POISON_PILL) {
              break;
            }
          }
        });
  }

  private static void sleepRandomTime() {
    ThreadLocalRandom random = ThreadLocalRandom.current();
    try {
      Thread.sleep(random.nextInt(MIN_SLEEP_TIME_MILLIS, MAX_SLEEP_TIME_MILLIS));
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
}

Code-Sprache: Java (java)

Hier die ersten 10 Zeilen einer beispielhaften Ausgabe:

[Thread-1] queue.offer(982)        --> queue = [982]
[Thread-6]     queue.poll() =  982 --> queue = []
[Thread-5]     queue.poll() = null --> queue = []
[Thread-0] queue.offer(917)        --> queue = [917]
[Thread-3] queue.offer(224)        --> queue = [917, 224]
[Thread-2] queue.offer(932)        --> queue = [917, 224, 932]
[Thread-6]     queue.poll() =  917 --> queue = [224, 932]
[Thread-4]     queue.poll() =  224 --> queue = [932]
[Thread-5]     queue.poll() =  932 --> queue = []
[Thread-0] queue.offer(607)        --> queue = [607]
[Thread-1] queue.offer( 87)        --> queue = [607, 87]
[Thread-3] queue.offer(264)        --> queue = [607, 87, 264]
[Thread-4]     queue.poll() =  607 --> queue = [87, 264]
[Thread-0] queue.offer(348)        --> queue = [87, 264, 348]
[Thread-2] queue.offer(728)        --> queue = [87, 264, 348, 728]
Code-Sprache: Klartext (plaintext)

Wir sehen sehr schön, wie die sieben Threads Elemente einfügen und entnehmen. In der dritten Zeile sehen wir auch, dass Thread 5 beim Aufruf von queue.poll() den Wert null geliefert bekommen hat, da die Queue zu diesem Zeitpunkt leer war.

ConcurrentLinkedQueue Performance

Dieser Abschnitt behandelt die Threadsicherheit und Zeitkomplexität von ConcurrentLinkedQueue.

Ist ConcurrentLinkedQueue threadsicher?

Die Threadsicherheit der ConcurrentLinkedQueue wird durch optimistisches Locking erzielt. Genauer gesagt: durch nicht-blockierende Compare-and-set (CAS)-Operationen auf separate VarHandles für den Kopf und das Ende der Queue.

Beim Zugriff auf Queues ist normalerweise niedrige bis moderate Contention (Zugriffskonflikte durch mehrere Threads) zu erwarten, da ein Thread in der Regel nicht ununterbrochen auf die Queue zugreift, sondern das einzustellende Element erstmal erzeugen bzw. das entnommende Element verarbeiten muss.

Bei niedriger bis moderater Contention erzielt optimistisches Locking einen deutlichen Performance-Gewinn gegenüber pessimistischem Locking durch implizite oder explizite Locks.

Unterschiede zu anderen Queues:

• Bei LinkedBlockingQueue wird die Threadsicherheit durch pessimistisches Locking via zweier ReentrantLocks gewährleistet, was zu besserer Performance bei hoher Contention führen kann.
• Bei ArrayBlockingQueue wird die Threadsicherheit durch ein einzelnes ReentrantLock gewährleistet.

ConcurrentLinkedQueue Zeitkomplexität

Wie bei allen Queues 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 hingegen nicht für die size()-Methode. Um die Länge der Queue zu bestimmen, muss über alle Elemente der verketteten Liste iteriert werden. Je länger die Queue, desto länger dauert es die Länge zu berechnen. Die Zeitkomplexität für size() beträgt somit O(n).

(Hier findest du eine Einführung in das Thema Zeitkomplexität.)

Zusammenfassung und Ausblick

In diesem Teil der Tutorialserie habe ich dir die konkrete Queue-Implementierung ConcurrentLinkedQueue und deren Eigenschaften vorgestellt.

Im folgenden Teil wird es um die PriorityQueue gehen, die einige Überraschungen bereithält.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

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 bspw. erzeugt eine auf 100 Elemente begrenzte ArrayBlockingQueue:

Queue<Integer> queue = new ArrayBlockingQueue<>(100);Code-Sprache: Java (java)

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 ander Thread ein Element entnehmen).
• Dequeue-Methoden, die beim Entnehmen eines Element 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.

Fairness Policy

Blockierende Methoden werden dabei nicht automatisch in der Reihenfolge bedient, in der sie aufgerufen wurden. Die Abarbeitung in Aufrufreihenfolge kann bei einigen Queue-Implementierungen durch eine optionale „Fairness Policy“ aktiviert werden. Diese erhöht allerdings den Overhead und verringert damit den Durchsatz der Queue massiv. In der Regel ist die Aktivierung der „Fairness Policy“ nicht nötig.

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 hinzugekommenden, blockierenden Methoden:

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 solange, bis Platz freigeworden 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 freigeworden 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() solange, 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.

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.

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();

  public static void main(String[] args) 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));
  }
}Code-Sprache: Java (java)

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ürzen 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 = [])Code-Sprache: Klartext (plaintext)

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, da diese mit den Elementen [4, 5, 6] ihre Kapazitätsgrenze von 3 erreicht hat.

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 Implementierungen des BlockingQueue-Interfaces mit jeweils spezifischen Eigenschaften. In dem folgenden UML-Klassendiagramm sind diese 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

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

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 und dessen Methoden put(), offer(), take() und poll() kennengelernt.

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

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

Das JDK enthält seit Java 5.0 das Interface java.util.Queue und mehrere Queue-Implementierungen, die sich in diversen Eigenschaften (wie bounded/unbounded, blocking/non-blocking, threadsicher/nicht threadsicher) unterscheiden.

Auf all diese Eigenschaften werde ich im weiteren Verlauf dieses Tutorials eingehen.

Java Queue Klassenhierarchie

Bevor ich die Java-Queue im Detail vorstelle, möchte ich einen Überblick in Form eines UML-Klassendiagramms geben:

Das BlockingQueue-Interface werde ich im nächsten Teil des Tutorials beschreiben.

Die konkreten Queue-Klassen ConcurrentLinkedQueue, PriorityQueue, ArrayBlockingQueue, DelayQueue, LinkedBlockingQueue, PriorityBlockingQueue und SynchronousQueue folgen im Anschluss. Das TransferQueue-Interface erkläre ich zusammen mit der LinkedTransferQueue.

Du kannst zu den entsprechenden Teilen jederzeit über die Tutorial-Navigation am rechten Rand springen.

Die grau eingezeichneten Interfaces Deque und BlockingDeque mitsamt ihren Implementierungen werden in der Tutorial-Serie „Deques“ behandelt.

Java Queue Methoden

Das Queue-Interface definiert sechs Methoden zum Einfügen, Entnehmen und Betrachten von Elementen. Für jede der drei Queue-Operationen „Enqueue“, „Dequeue“ und „Peek“ definiert das Interface jeweils zwei Methoden: eine die im Fehlerfall eine Exception wirft und eine die einen speziellen Wert (false oder null) zurückliefert.

Methoden zum Einfügen in die Queue

Zunächst ein grafischer Überblick über die Enqueue-Methoden:

Queue.add()

Diese Methode ist bereits im Collection-Interface definiert und fügt ein Element in die Queue ein. Bei Erfolg gibt die Methode true zurück. Wenn eine größenbeschränkte Queue voll ist, wirft diese Methode eine IllegalStateException.

Queue.offer()

offer() fügt wie add() ein Element in die Queue ein und gibt bei Erfolg true zurück. Wenn eine größenbeschränkte Queue voll ist, gibt diese Methode false zurück anstatt eine IllegalStateException zu werfen.

Methoden zum Entnehmen aus der Queue

Auch für die Dequeue-Methoden zunächst ein grafischer Überblick:

Queue.remove()

remove() entnimmt das Element vom Kopf der Queue. Ist die Queue leer, wirft die Methode eine NoSuchElementException.

Queue.poll()

Auch poll() entnimmt das Element am Kopf der Queue. Anders als remove() wirft die Methode bei einer leeren Queue keine Exception, sondern gibt null zurück.

Methoden zum Betrachten des Kopf-Elements

Und wieder zunächst ein Überblick über Methoden:

Queue.element()

Die element()-Methode gibt das Element vom Kopf der Queue zurück, ohne es aus der Queue zu entnehmen. Ist die Queue leer, wird eine NoSuchElementException geworfen.

Queue.peek()

Genau wie element() gibt auch peek() das Kopf-Element zurück, ohne es aus der Queue zu entfernen. Bei einer leeren Queue gibt diese Methode allerdings – genau wie poll() – null zurück.

Queue-Methoden – Zusammenfassung

Die folgende Tabelle zeigt noch einmal die sechs Methoden gruppiert nach Operation und Art der Fehlerbehandlung:

Wie erzeugt man eine Queue?

java.util.Queue ist ein Interface. Ein Interface kann nicht instatiiert werden, da es lediglich beschreibt, welche Methoden eine Klasse anbietet, jedoch keine Implementierungen dieser Methoden beinhaltet.

Was passiert, wenn man es dennoch versucht?

public class QueueTest {
  public static void main(String[] args) {
    Queue<Integer> queue = new Queue<>(); // <-- Don't do this!
  }
}Code-Sprache: Java (java)

Beim Versuch diesen Code zu compilieren würdest du folgende Fehlermeldung sehen:

QueueTest.java:5: error: Queue is abstract; cannot be instantiated
    Queue<Integer> queue = new Queue<>(); // <-- Don't do this!
                           ^
1 errorCode-Sprache: Klartext (plaintext)

Daher muss eine der konkreten Queue-Implementierungen ausgewählt werden, z. B. die ConcurrentLinkedQueue:

Queue<Integer> queue = new ConcurrentLinkedQueue<>();Code-Sprache: Java (java)

(Die verschiedenen Queue-Klassen werden in weiteren Teilen des Tutorials erläutert. Im letzten Teil findest du eine Entscheidungshilfe, wann du welche Implementierung verwenden solltest.)

Beispiel: Wie benutzt man eine Queue?

Das folgende Beispiel zeigt, wie man eine Queue erstellt, wie man diese mit einigen Werten befüllt, und wie man die Werte wieder entnimmt. Du findest den Beispiel-Code auch auf GitHub.

public class JavaQueueDemo {
  public static void main(String[] args) {
    // 1.
    Queue<Integer> queue = new ConcurrentLinkedQueue<>();

    // 2.
    for (int i = 1; i <= 5; i++) {
      queue.offer(i);
      System.out.println("queue.offer(" + i + ") --> queue = " + queue);
    }

    System.out.println();

    // 3.
    System.out.println("queue.peek() = " + queue.peek());

    System.out.println();

    // 4.
    while (!queue.isEmpty()) {
      System.out.println("queue.poll() = " + queue.poll() + " --> queue = " + queue);
    }

    System.out.println();

    // 5.
    System.out.println("queue.poll() = " + queue.poll());
    System.out.println("queue.peek() = " + queue.peek());
  }
}
Code-Sprache: Java (java)

Das Programm tut folgendes (die Numerierung verweist auf die Kommentare im Quellcode):

1. Es erstellt eine Queue. Welche du benutzt, ist für dieses Beispiel irrelevant, da es keine speziellen Queue-Eigenschaften erfordert. Wir verwenden die ConcurrentLinkedQueue.
2. Die Werte 1 bis 5 werden mit Queue.offer() in die Queue geschrieben. Der Inhalt der Queue wird nach jedem Einfügen angezeigt.
3. Wir betrachten mit Queue.peek() das Kopf-Element der Queue.
4. Solange die Queue Elemente enthält (dies prüfen wir mit der isEmpty()-Methode, die das Queue-Interface von Collection erbt), werden diese mit Queue.poll() entnommen und angezeigt. Danach wird jeweils wieder der gesamte Inhalt der Queue angezeigt.
5. Nachdem die Queue geleert wurde, werden noch einmal die Rückgabewerte von poll() und peek() angezeigt.

Das Programm gibt folgendes aus:

queue.offer(1) --> queue = [1]
queue.offer(2) --> queue = [1, 2]
queue.offer(3) --> queue = [1, 2, 3]
queue.offer(4) --> queue = [1, 2, 3, 4]
queue.offer(5) --> queue = [1, 2, 3, 4, 5]

queue.peek() = 1

queue.poll() = 1 --> queue = [2, 3, 4, 5]
queue.poll() = 2 --> queue = [3, 4, 5]
queue.poll() = 3 --> queue = [4, 5]
queue.poll() = 4 --> queue = [5]
queue.poll() = 5 --> queue = []

queue.poll() = null
queue.peek() = nullCode-Sprache: Klartext (plaintext)

Es ist sehr gut zu sehen, wie die Elemente in derselben Reihenfolge entnommen werden, wie sie eingefügt wurden (First-in-First-out – FIFO).

Zusammenfassung und Ausblick

In diesem Teil des Tutorials hast du das Queue-Interface von Java kennengelernt. Anhand eines Beispiels hast du gesehen, wie man die Java-Queue benutzt.

Im nächsten Teil schauen wir uns das Interface „BlockingQueue“ genauer an. Dort werde ich auch den Unterschied zwischen bounded und unbounded bzw. blocking und non-blocking Queues erklären.

Danach werden wir alle Queue-Implementierungen des JDK einzeln betrachten. Anhand deren besonderer Eigenschaften werde ich dir erklären, wann man welche Implementierung einsetzen sollte.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.

In diesem Tutorial lernst du alles über den abstrakten Datentyp „Queue“ (im deutschen auch als „Warteschlange“ bezeichnet):

• Wie funktioniert eine Queue?
• Was sind die Anwendungsgebiete für Queues?
• Welche Queue-Interfaces und -Klassen gibt es im JDK?
• Was sind blocking, non-blocking, bounded und unbounded Queues?
• Wie implementiert man eine Queue in Java?

Was ist eine Queue?

Eine Queue ist eine Liste von Elementen, bei der die Elemente auf einer Seite eingefügt und in derselben Reihenfolge auf der anderen Seite wieder entnommen werden.

Man kann sich das wie eine Warteschlange an einer Kasse oder bei einer Behörde vorstellen:

Ankommende Personen stellen sich am Ende der Schlange (rechts im Bild) an. Sobald ein Kunde abgefertigt wurde, kommt der nächste Kunde vom Kopf der Schlange (links) an die Reihe.

Die Person, die sich zuerst angestellt hat, kommt somit auch zuerst an die Reihe. Daher sprechen wir vom First-in-First-out-Prinzip (FIFO).

FIFO-Prinzip bei Queues

Beim abstrakten Datentyp „Queue“ kann das wie im folgenden Beispiel aussehen:

Die Grafik zeigt eine Queue, die die Elemente 6, 7, 8, usw. bis 13 enhält. Die 5 wurde gerade vom Kopf der Queue (englisch „Front“ oder „Head“, links im Bild) entnommen und die 14 wird gerade an das Ende der Queue („Back“, „Tail“ oder „Rear“ im englischen, rechts im Bild) eingefügt.

Queue Operationen: Enqueue und Dequeue

Die Operationen der Queue bezeichnen wir wie folgt:

• „Enqueue“: Einfügen neuer Elemente am Ende der Queue
• „Dequeue“: Entnehmen von Elementen vom Kopf der Queue
• „Peek“ oder „Front“: Betrachten des Elements am Kopf, ohne es zu entnehmen (optional)

(Die entsprechenden Methoden der Java-Queue-Implementierungen heißen übrigens anders; mehr dazu im nächsten Teil des Tutorials, „Java Queue-Interface„.)

Anwendungsgebiete für Queues

Ein Anwendungsbereich von Queues, den wir alle kennen, ist die Druckerwarteschlange. Verschiedene Programme stellen dort Druckaufträge ein. In der Regel gibt es nur einen Drucker, der diese dann der Reihe nach abarbeitet.

Ein technisches Anwendungsbeispiel ist die Verarbeitung von HTTP-Requests in einem Webserver. Ein Webserver arbeitet in der Regel mit einem Threadpool für gleichzeitig abzuarbeitende Requests. Wenn mehr Anfragen hereinkommen als gleichzeitig abgearbeitet werden können, ist der Threadpool ausgelastet. Zusätzliche Requests werden dann in eine Warteschlange gestellt und in First-in-First-out-Reihenfolge abgearbeitet, sobald wieder Threads zur Verfügung stehen.

Zeitkomplexität der Queue-Operationen

Eine Einführung in das Thema Zeitkomplexität findest du im Artikel „O-Notation und Zeitkomplexität – anschaulich erklärt„.

Queues werden in der Regel mit Arrays oder verketteten Listen implementiert. Bei beiden Varianten ist der Aufwand für die Enqueue- und Dequeue-Operationen jeweils konstant, d. h. der Aufwand ändert sich nicht mit der Länge der Queue.

Die Zeitkomplexität dieser Operationen lautet also: O(1).

Zu Übungszwecken kann man eine Queue auch mit Stacks implementieren (dazu mehr in einem späteren Teil des Tutorials). Die Zeitkomplexität ist dann allerdings höher.

Wenn du noch Fragen hast, stelle sie gerne über die Kommentar-Funktion. Möchtest du über neue Tutorials und Artikel informiert werden? Dann klicke hier, um dich für den HappyCoders.eu-Newsletter anzumelden.