DelayQueue in Java: Entnahme nach Wartezeit (mit Beispiel)

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 jedes Element erst herausgibt, wenn seine Wartezeit abgelaufen ist.

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 erst dann entnommen, wenn eine diesem Element zugeordnete Wartezeit („Delay“) abgelaufen ist.

Dazu müssen die Elemente das Interface java.util.concurrent.Delayed implementieren. Dessen Methode getDelay() liefert bei jedem Aufruf die Wartezeit, die noch ablaufen muss, bis das Element aus der Queue entnommen werden darf. Ist sie null oder negativ, ist das Element abgelaufen.

Delayed erweitert außerdem Comparable<Delayed>, und die Javadoc von Delayed verlangt, dass compareTo() die Elemente in derselben Reihenfolge ordnet wie getDelay(). Denn die PriorityQueue sortiert nach compareTo(), und die DelayQueue prüft nur beim Element am Kopf, ob es abgelaufen ist. Ordnet compareTo() anders, liegt am Kopf ein Element, dessen Wartezeit noch läuft, während ein abgelaufenes dahinter wartet.

DelayQueue ist blockierend, aber unbounded: Sie kann beliebig viele Elemente aufnehmen und blockiert nur bei der Entnahme (bis die Wartezeit des Elements am Kopf abgelaufen ist), nicht beim Einfügen. null als Element ist nicht erlaubt.

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

Warten mehrere Threads in take(), wartet nur einer von ihnen – der sogenannte Leader – mit Zeitlimit, nämlich bis zum Ablauf des Elements am Kopf. Leader wird der Thread, der als erster einen nicht abgelaufenen Kopf vorfindet, solange es noch keinen Leader gibt. Alle weiteren Threads warten ohne Zeitlimit.

Geweckt wird von ihnen immer nur einer, und zwar in zwei Fällen:

1. Der Leader hat das abgelaufene Element entnommen. Bevor er take() verlässt, weckt er einen wartenden Thread, sofern die Queue noch Elemente enthält. Dieser Thread prüft das neue Element am Kopf: Ist es schon abgelaufen, entnimmt er es und weckt seinerseits den nächsten Thread. Sonst wird er der neue Leader.
2. offer() fügt ein Element ein, das früher abläuft als der bisherige Kopf, oder die Queue war leer. Dann setzt offer() den Leader zurück und weckt einen wartenden Thread – nicht unbedingt den bisherigen Leader. Dieser wird mit der kürzeren Wartezeit neuer Leader. Der bisherige Leader wartet bis zum Ende seiner alten Frist. Findet er dann kein abgelaufenes Element am Kopf, wartet er wie die übrigen Threads ohne Zeitlimit.

So wacht beim Ablauf eines Elements nur ein Thread auf, nicht alle wartenden.

Die Eigenschaften im Detail:

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

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

Abgelaufene und nicht abgelaufene Elemente

Die DelayQueue implementiert das BlockingQueue-Interface, verletzt allerdings absichtlich dessen allgemeinen Vertrag: Laut Queue und BlockingQueue geben die Methoden zum Entnehmen das Element am Kopf zurück. null liefern, eine Exception werfen oder warten dürfen sie nur, wenn die Queue leer ist. Bei der DelayQueue tun sie das auch dann, wenn die Queue Elemente enthält, aber keines davon abgelaufen ist.

Die folgenden Methoden berücksichtigen also nur abgelaufene Elemente und entnehmen immer nur das abgelaufene Element am Kopf:

• poll() liefert null, solange kein Element abgelaufen ist – laut Interface nur bei leerer Queue.
• remove() ohne Argument wirft eine NoSuchElementException, solange kein Element abgelaufen ist – laut Interface nur bei leerer Queue.
• take() wartet, bis das erste Element abgelaufen ist – laut Interface nur, bis ein Element verfügbar ist.
• poll(long, TimeUnit) wartet ebenso und liefert null, wenn innerhalb der Wartezeit kein Element abläuft.
• drainTo() überträgt nur die abgelaufenen Elemente. Das deckt sich mit dem Interface, das „alle verfügbaren“ Elemente verspricht.

Alle anderen Methoden arbeiten auf allen Elementen, ob abgelaufen oder nicht:

• size() zählt alle Elemente.
• peek() liefert das Element am Kopf, auch wenn seine Wartezeit noch läuft – also das Element, das als Nächstes abläuft.
• Der Iterator durchläuft alle Elemente in keiner bestimmten Reihenfolge. Er arbeitet auf einer Kopie, die beim Erzeugen des Iterators angelegt wird, und bildet spätere Änderungen also nicht ab.
• remove(Object) und clear() entfernen auch Elemente, deren Wartezeit noch läuft.

Eine Schleife, die isEmpty() abfragt und dann poll() aufruft, bekommt deshalb null, obwohl die Queue nicht leer ist. Das Beispiel im Abschnitt „DelayQueue Beispiel“ ruft daher take() auf.

Einsatzempfehlung

Für Aufgaben, die nach einer Wartezeit oder in festen Abständen laufen sollen, empfehle ich einen ScheduledExecutorService, z. B. aus Executors.newScheduledThreadPool(). Er arbeitet nach demselben Prinzip: Der ScheduledThreadPoolExecutor hält seine Aufgaben in einer eigenen DelayedWorkQueue, einem Heap, sortiert nach Ausführungszeitpunkt. Zusätzlich verwaltet er die Threads, die die Aufgaben ausführen, wiederholt periodische Aufgaben und liefert dir für jede Aufgabe ein ScheduledFuture.

Die DelayQueue selbst setzt du ein, wenn deine Elemente keine Aufgaben sind, sondern Daten, die deine eigenen Threads nach Ablauf ihrer Wartezeit abholen – etwa Cache-Einträge, die nach einer Frist entfernt werden sollen, oder Aufträge, die nach einem Fehlschlag erst nach einer Wartezeit erneut versucht werden sollen.

Im JDK selbst verwendet der Swing-Timer eine DelayQueue: Jeder gestartete javax.swing.Timer wird, verpackt in einen DelayedTimer, in die DelayQueue der Klasse javax.swing.TimerQueue eingefügt, und ein einziger Thread entnimmt die Timer dort mit take().

Alle Operationen laufen allerdings über dasselbe ReentrantLock – bei hoher Contention warten Producer und Consumer aufeinander. Und Elemente, die Delayed nicht selbst implementieren, brauchen eine Wrapper-Klasse wie DelayedElement im folgenden Beispiel.

DelayQueue Beispiel

Im folgenden Beispiel (→ Code auf GitHub) wird eine DelayQueue mit sieben Instanzen der Klasse DelayedElement gefüllt, die jeweils eine Zufallszahl enthalten sowie ein zufälliges initiales Delay zwischen 100 und 999 ms. Danach wird so oft take() aufgerufen, bis die Queue wieder leer ist.

public class DelayQueueExample {
  static void main() {
    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.take() = %s --> queue = %s%n",
            System.currentTimeMillis() - startTime, element, queue);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }
}

Die zugehörige Klasse DelayedElement (→ Code auf GitHub) berechnet im Konstruktor den Ablaufzeitpunkt expirationNanos. getDelay() liefert die Differenz zwischen diesem Zeitpunkt und der aktuellen Zeit, umgerechnet in die angefragte Einheit. compareTo() vergleicht die Ablaufzeitpunkte und ordnet die Elemente damit in derselben Reihenfolge wie getDelay().

Die Klasse rechnet mit System.nanoTime(), nicht mit System.currentTimeMillis(). currentTimeMillis() liefert die Uhrzeit des Systems, und die springt, wenn sie z. B. per NTP korrigiert wird – ein Element liefe dann zu früh oder zu spät ab. nanoTime() misst dagegen nur verstrichene Zeit; die Delayed-Implementierungen des JDK im ScheduledThreadPoolExecutor und in der TimerQueue verwenden es ebenso.

Zwei Werte von nanoTime() vergleicht compareTo() über ihre Differenz, nicht direkt. So verlangt es die Javadoc von nanoTime(), weil die Werte überlaufen können.

Hier die Klasse DelayedElement:

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class DelayedElement<E> implements Delayed {
  private final E e;
  private final long initialDelayMillis;
  private final long expirationNanos;

  public DelayedElement(E e, long initialDelayMillis) {
    this.e = e;
    this.initialDelayMillis = initialDelayMillis;
    this.expirationNanos =
        System.nanoTime() + MILLISECONDS.toNanos(initialDelayMillis);
  }

  @Override
  public long getDelay(TimeUnit unit) {
    long remainingNanos = expirationNanos - System.nanoTime();
    return unit.convert(remainingNanos, NANOSECONDS);
  }

  @Override
  public int compareTo(Delayed o) {
    DelayedElement<?> other = (DelayedElement<?>) o;
    // Compare the difference, not the values: System.nanoTime() may overflow
    return Long.signum(expirationNanos - other.expirationNanos);
  }

  @Override
  public String toString() {
    return "{%s, %dms}".formatted(e, initialDelayMillis);
  }
}

Die folgende Ausgabe eines Laufs unter Java 27 zeigt, dass die Entnahme der Ablaufzeit folgt, nicht der Reihenfolge des Einfügens:

[  1ms] queue.offer({91, 560ms})   --> queue = [{91, 560ms}]
[ 22ms] queue.offer({46, 169ms})   --> queue = [{46, 169ms}, {91, 560ms}]
[ 23ms] queue.offer({94, 476ms})   --> queue = [{46, 169ms}, {91, 560ms}, {94, 476ms}]
[ 23ms] queue.offer({69, 951ms})   --> queue = [{46, 169ms}, {91, 560ms}, {94, 476ms}, {69, 951ms}]
[ 23ms] queue.offer({23, 613ms})   --> queue = [{46, 169ms}, {91, 560ms}, {94, 476ms}, {69, 951ms}, {23, 613ms}]
[ 23ms] queue.offer({67, 141ms})   --> queue = [{67, 141ms}, {91, 560ms}, {46, 169ms}, {69, 951ms}, {23, 613ms}, {94, 476ms}]
[ 23ms] queue.offer({95, 170ms})   --> queue = [{67, 141ms}, {91, 560ms}, {46, 169ms}, {69, 951ms}, {23, 613ms}, {94, 476ms}, {95, 170ms}]
[166ms] queue.take() = {67, 141ms} --> queue = [{46, 169ms}, {91, 560ms}, {95, 170ms}, {69, 951ms}, {23, 613ms}, {94, 476ms}]
[196ms] queue.take() = {46, 169ms} --> queue = [{95, 170ms}, {91, 560ms}, {94, 476ms}, {69, 951ms}, {23, 613ms}]
[197ms] queue.take() = {95, 170ms} --> queue = [{94, 476ms}, {91, 560ms}, {23, 613ms}, {69, 951ms}]
[500ms] queue.take() = {94, 476ms} --> queue = [{91, 560ms}, {69, 951ms}, {23, 613ms}]
[562ms] queue.take() = {91, 560ms} --> queue = [{23, 613ms}, {69, 951ms}]
[639ms] queue.take() = {23, 613ms} --> queue = [{69, 951ms}]
[979ms] queue.take() = {69, 951ms} --> queue = []

Die Queue in der Ausgabe ist nicht sortiert², das Element mit der kürzesten verbleibenden Wartezeit steht aber immer vorne (links). Und take() kehrt jeweils wenige Millisekunden zurück, nachdem die Wartezeit des Elements abgelaufen ist: Das Element {67, 141ms} wurde bei 23 ms eingefügt, lief also bei 164 ms ab und wurde bei 166 ms entnommen.

² Tatsächlich siehst du die Reihenfolge der Elemente in der Array-Repräsentation des Min-Heaps.

Das folgende Diagramm zeigt denselben Lauf im Zeitverlauf. Jede Zeile steht für ein Element, in der Reihenfolge des Einfügens; der Balken reicht von offer() bis take(), und an seinem Ende steht, als wievieltes das Element entnommen wurde:

Das Element {91, 560ms} wurde als erstes eingefügt und als fünftes entnommen, das Element {67, 141ms} als sechstes eingefügt und als erstes entnommen.

Zusammenfassung und Ausblick

In diesem Artikel hast du die DelayQueue kennengelernt. Sie gibt jedes Element erst nach Ablauf seiner Wartezeit heraus, ihre Elemente implementieren Delayed mit einem compareTo(), das zu getDelay() passt, und poll() und take() berücksichtigen nur abgelaufene Elemente. Für Aufgaben, die nach einer Wartezeit laufen sollen, ist ein ScheduledExecutorService die einfachere Wahl.

Einen Vergleich aller Queue-Implementierungen des JDK und eine Übersicht, in welchen Fällen du welche Implementierung verwenden solltest, findest du im Artikel über das Queue-Interface.

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