In diesem Artikel erfährst du, wie die PriorityBlockingQueue funktioniert und welche Eigenschaften sie hat. Anhand von Beispielen siehst du, wie man sie einsetzt – und wie sie als Work Queue eines ThreadPoolExecutors Aufgaben nach Priorität abarbeitet.
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 Priority Queue ist.
Wie bei der PriorityQueue werden die Elemente in einem Array gespeichert, das einen Min-Heap repräsentiert. Am Kopf der Queue steht also immer das kleinste Element.
Der Iterator liefert die Elemente dagegen nicht in Prioritätsreihenfolge, sondern in der Reihenfolge des internen Heap-Arrays – dasselbe gilt für toString() und toArray(). Sortiert entnimmst du sie mit take(), poll() oder drainTo().
Threadsicherheit wird durch ein einzelnes ReentrantLock sichergestellt (s. u.).
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.
Ihre einzige Grenze ist der Arbeitsspeicher: Reicht er zum Vergrößern des Arrays nicht mehr aus, kann das Einfügen laut Javadoc mit einem OutOfMemoryError scheitern.
null-Elemente sind nicht erlaubt (put(null) wirft eine NullPointerException).
Die Eigenschaften im Detail:
| Unterliegende Datenstruktur | Thread-safe? | Blocking/ Non-blocking | Fairness Policy | Bounded/ Unbounded | Iterator Type |
|---|---|---|---|---|---|
| Min-Heap (gespeichert in einem Array) | Ja (pessimistisches Locking mit einem Lock) | Blocking (nur dequeue) | Nicht verfügbar | Unbounded | Weakly consistent¹ |
¹ Weakly consistent: Laut Javadoc kann der Iterator Änderungen nach seiner Erzeugung berücksichtigen, muss es aber nicht. Tatsächlich berücksichtigt er keine: Er arbeitet auf einer Kopie des Arrays, die iterator() beim Aufruf anlegt, liefert jedes Element, das in diesem Moment in der Queue liegt, genau einmal und wirft nie eine ConcurrentModificationException.
Der Iterator unterstützt remove(). Die Methode entfernt das Element aus der Queue selbst, nicht nur aus der Kopie.
Einsatzempfehlung
Ich empfehle die PriorityBlockingQueue, wenn mehrere Threads Elemente nach Priorität austauschen statt in der Reihenfolge des Einfügens – etwa Aufgaben, die ein Thread-Pool nach Dringlichkeit abarbeiten soll. Wie das mit einem ThreadPoolExecutor geht, zeigt der Abschnitt „PriorityBlockingQueue im ThreadPoolExecutor“.
Alle Operationen laufen allerdings über dasselbe ReentrantLock – bei hoher Contention (vielen Threads, die gleichzeitig auf die Queue zugreifen) warten Producer und Consumer aufeinander.
Und weil die Queue unbounded ist, bremst sie keinen Producer. Kommen Elemente schneller hinein, als sie entnommen werden, wächst die Queue, bis der Arbeitsspeicher ausgeht.
Brauchst du keine Priorität, sondern die Reihenfolge des Einfügens, ist die LinkedBlockingQueue mit ihren zwei Locks meist die bessere Wahl; brauchst du eine Kapazitätsgrenze, die ArrayBlockingQueue.
PriorityBlockingQueue Beispiel
Das folgende 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();
static void main() 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 second (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 e) {
Thread.currentThread().interrupt();
}
}
private static void dequeue(BlockingQueue<Integer> queue) {
log(" Calling queue.take() (queue = %s)...", queue);
try {
Integer element = queue.take();
log(" queue.take() returned %d (queue = %s)", element, queue);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private static void log(String format, Object... args) {
System.out.printf(
Locale.US,
"[%4.1fs] [%-16s] %s%n",
(System.currentTimeMillis() - startTime) / 1000.0,
Thread.currentThread().getName(),
String.format(format, args));
}
}Im Folgenden siehst du 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 = [])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.
Dahinter steht nach 8,5 s die 71 vor der 33. Das liegt daran, dass toString() das Heap-Array ausgibt – und das Heap-Array ist nur am Kopf sortiert. Wie Baum und Array zusammenhängen, zeigt das Diagramm am Ende dieses Abschnitts.
Nach 9,0 s wird das kleinste Element, 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.
Nach 12,0 s entnimmt Thread 9 die 19, und die Queue gibt [33, 71, 58] aus – auch hier steht die 71 vor der kleineren 58.
Das folgende Diagramm zeigt den Heap vor und nach diesem take(), jeweils als Baum und als Array. Das Array enthält den Baum Ebene für Ebene von links nach rechts.
Beim Entnehmen holt die Queue das letzte Element des Arrays, die 71, an die Wurzel. Weil die 71 größer ist als ihre beiden Kinder 33 und 58, tauscht sie den Platz mit dem kleineren der beiden, der 33 – danach steht wieder das kleinste Element an der Wurzel:
![Zwei Zustände des Min-Heaps als Baum und als Array: nach put(19) die 19 an der Wurzel mit den Kindern 33 und 58 und der 71 unter der 33, Array [19, 33, 58, 71]; nach take() die 33 an der Wurzel mit den Kindern 71 und 58, Array [33, 71, 58]](/_astro/priorityblockingqueue-heap-vor-und-nach-take-de.ldS_Qa66_Z2af4FL.svg)
Wie dieses sogenannte Sift Down Schritt für Schritt abläuft, zeigt der Artikel „Priority Queue mit einem Min-Heap implementieren“.
PriorityBlockingQueue mit Comparator
Ohne weitere Angabe ordnet die PriorityBlockingQueue ihre Elemente nach ihrer natürlichen Ordnung, also nach compareTo(). Die Elemente müssen dann Comparable implementieren. Ein Element, das das nicht tut, weist die Queue beim Einfügen mit einer ClassCastException zurück.
Eine andere Reihenfolge legst du mit einem Comparator fest. Von den vier Konstruktoren der PriorityBlockingQueue nimmt nur einer einen Comparator entgegen, und dieser Konstruktor verlangt zusätzlich eine Anfangskapazität.
Das folgende Programm legt eine Queue mit Anfangskapazität 11 und Comparator.reverseOrder() an, schreibt drei Zahlen hinein und entnimmt sie wieder (→ Code auf GitHub):
public class PriorityBlockingQueueComparatorExample {
static void main() throws InterruptedException {
BlockingQueue<Integer> queue =
new PriorityBlockingQueue<>(11, Comparator.reverseOrder());
queue.put(31);
queue.put(87);
queue.put(15);
IO.println(queue.take() + " " + queue.take() + " " + queue.take());
}
}Das Programm gibt Folgendes aus:
87 31 15Mit Comparator.reverseOrder() steht das größte Element am Kopf – aus dem Min-Heap wird ein Max-Heap.
Die übrigen drei Konstruktoren nehmen keinen Comparator entgegen:
- Der Konstruktor ohne Parameter legt das Array mit 11 Plätzen an.
- Der Konstruktor mit einem
intlegt es mit der übergebenen Anfangskapazität an. Eine Anfangskapazität unter 1 weist er – wie der Konstruktor mitComparator– mit einerIllegalArgumentExceptionzurück. - Der Konstruktor mit einer Collection übernimmt deren Elemente. Ist sie ein
SortedSetoder einePriorityBlockingQueue, übernimmt er auch deren Ordnung statt der natürlichen.
Elemente mit gleicher Priorität
In welcher Reihenfolge die Queue Elemente gleicher Priorität herausgibt, legt sie nicht fest. Auch die Reihenfolge des Einfügens ist nicht garantiert.
Brauchst du diese Reihenfolge, gibst du jedem Element eine fortlaufende Nummer als zweiten Sortierschlüssel. Die Javadoc der Klasse zeigt das mit der Klasse FIFOEntry: Sie zieht die Nummer beim Erzeugen aus einem AtomicLong und vergleicht sie, wenn die Elemente selbst gleich sind.
PriorityBlockingQueue im ThreadPoolExecutor
Ein ThreadPoolExecutor nimmt seine Work Queue als BlockingQueue im Konstruktor entgegen. Mit einer PriorityBlockingQueue arbeitet er wartende Aufgaben nach Priorität ab statt in der Reihenfolge, in der sie eingereicht wurden.
Das folgende Programm reicht vier Aufgaben vom Typ PrioritizedTask an einen Pool mit nur einem Thread ein. Die erste Aufgabe, E, läuft sofort. Die übrigen drei warten in der Queue, und der Thread holt sie nach ihrer Priorität heraus – die kleinste Zahl zuerst (→ Code auf GitHub: PrioritizedTask, PriorityThreadPoolExample):
public record PrioritizedTask(int priority, String name)
implements Runnable, Comparable<PrioritizedTask> {
@Override
public void run() {
IO.println("Running task " + name + " (priority " + priority + ")");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@Override
public int compareTo(PrioritizedTask other) {
return Integer.compare(priority, other.priority);
}
}
public class PriorityThreadPoolExample {
static void main() throws InterruptedException {
ThreadPoolExecutor pool = new ThreadPoolExecutor(
1, 1, 0, TimeUnit.SECONDS, new PriorityBlockingQueue<>());
pool.execute(new PrioritizedTask(5, "E")); // runs at once on the only thread
pool.execute(new PrioritizedTask(3, "C")); // these three wait in the queue
pool.execute(new PrioritizedTask(1, "A"));
pool.execute(new PrioritizedTask(2, "B"));
pool.shutdown();
pool.awaitTermination(1, TimeUnit.MINUTES);
}
}Der Konstruktor des ThreadPoolExecutors bekommt im Listing fünf Argumente:
corePoolSize= 1: So viele Threads hält der Pool dauerhaft, auch wenn sie gerade nichts zu tun haben.maximumPoolSize= 1: Bis zu so vielen Threads darf der Pool wachsen.keepAliveTime= 0 undunit=TimeUnit.SECONDS: So lange wartet ein Thread übercorePoolSizehinaus auf neue Arbeit, bevor er endet.workQueue: diePriorityBlockingQueue, in der Aufgaben warten, solange kein Thread frei ist.
Das sleep() in run() hält den Thread lange genug bei Aufgabe E, damit C, A und B in der Queue ankommen, bevor er die nächste Aufgabe holt.
Das Programm gibt Folgendes aus:
Running task E (priority 5)
Running task A (priority 1)
Running task B (priority 2)
Running task C (priority 3)C wurde vor A und B eingereicht und läuft trotzdem als letzte Aufgabe.
Dass corePoolSize und maximumPoolSize im Listing gleich sind, hat einen Grund: Zusätzliche Threads über corePoolSize hinaus startet der ThreadPoolExecutor nur, wenn die Queue eine Aufgabe ablehnt – und eine unbounded Queue wie die PriorityBlockingQueue lehnt keine ab. Ein größerer Wert für maximumPoolSize bliebe also wirkungslos; so steht es auch in der Javadoc von ThreadPoolExecutor.
Ich empfehle daher, mit einer PriorityBlockingQueue beide Werte gleich zu setzen.
ClassCastException bei submit()
Ersetzt du im Listing execute() durch submit(), bricht das Programm schon beim zweiten Aufruf ab:
pool.submit(new PrioritizedTask(5, "E"));
pool.submit(new PrioritizedTask(3, "C"));Der zweite Aufruf wirft die folgende Exception:
Exception in thread "main" java.lang.ClassCastException: class java.util.concurrent.FutureTask cannot be cast to class java.lang.Comparable (java.util.concurrent.FutureTask and java.lang.Comparable are in module java.base of loader 'bootstrap')
at java.base/java.util.concurrent.PriorityBlockingQueue.siftUpComparable(PriorityBlockingQueue.java:350)
at java.base/java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:476)
at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1320)
at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:125)Der Grund: submit() verpackt die Aufgabe in einen FutureTask, den es als Future zurückgibt, und reicht diesen an execute() weiter. FutureTask implementiert Comparable nicht, und die Queue scheitert beim Einsortieren am Cast. Der erste Aufruf geht noch durch, weil der Pool die erste Aufgabe direkt seinem neuen Thread übergibt, ohne die Queue zu benutzen.
Brauchst du kein Future, reichst du die Aufgaben mit execute() ein wie im Listing oben. Brauchst du eines – etwa, um eine wartende Aufgabe abzubrechen oder auf ihr Ergebnis zu warten –, überschreibst du in einer Unterklasse von ThreadPoolExecutor die Methode newTaskFor(), die submit() den FutureTask liefert. Diese Möglichkeit sieht die Javadoc von AbstractExecutorService ausdrücklich vor.
Das folgende Listing enthält neben dem Record PrioritizedTask von oben drei neue Klassen (→ Code auf GitHub: PrioritizedFutureTask, PriorityThreadPoolExecutor, PriorityThreadPoolSubmitExample):
PrioritizedFutureTaskist einFutureTask, derComparableimplementiert und dazu die Priorität seiner Aufgabe vergleicht.PriorityThreadPoolExecutorliefert einen solchenPrioritizedFutureTaskausnewTaskFor().PriorityThreadPoolSubmitExamplereicht die Aufgaben mitsubmit()ein und bricht B über seinFutureab, solange B noch in der Queue wartet.
public record PrioritizedTask(int priority, String name)
implements Runnable, Comparable<PrioritizedTask> {
@Override
public void run() {
IO.println("Running task " + name + " (priority " + priority + ")");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@Override
public int compareTo(PrioritizedTask other) {
return Integer.compare(priority, other.priority);
}
}
public class PrioritizedFutureTask<T> extends FutureTask<T>
implements Comparable<PrioritizedFutureTask<?>> {
private final PrioritizedTask task;
PrioritizedFutureTask(PrioritizedTask task, T value) {
super(task, value);
this.task = task;
}
@Override
public int compareTo(PrioritizedFutureTask<?> other) {
return task.compareTo(other.task);
}
}
public class PriorityThreadPoolExecutor extends ThreadPoolExecutor {
PriorityThreadPoolExecutor(int poolSize) {
super(poolSize, poolSize, 0, TimeUnit.SECONDS, new PriorityBlockingQueue<>());
}
@Override
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new PrioritizedFutureTask<>((PrioritizedTask) runnable, value);
}
}
public class PriorityThreadPoolSubmitExample {
static void main() throws InterruptedException {
ThreadPoolExecutor pool = new PriorityThreadPoolExecutor(1);
pool.submit(new PrioritizedTask(5, "E"));
pool.submit(new PrioritizedTask(3, "C"));
pool.submit(new PrioritizedTask(1, "A"));
Future<?> taskB = pool.submit(new PrioritizedTask(2, "B"));
taskB.cancel(false); // B is still waiting in the queue
pool.shutdown();
pool.awaitTermination(1, TimeUnit.MINUTES);
}
}Das Programm gibt nur noch drei Zeilen aus – B fehlt, und die übrigen Aufgaben laufen weiter nach Priorität:
Running task E (priority 5)
Running task A (priority 1)
Running task C (priority 3)Die abgebrochene Aufgabe bleibt in der Queue, bis der Thread sie herausholt. Ausgeführt wird sie dann nicht mehr, so steht es in der Javadoc von ThreadPoolExecutor. Den Grund zeigt der Quelltext von FutureTask.run(): Die Methode kehrt sofort zurück, weil der Task nicht mehr in seinem Ausgangszustand ist.
Der Cast in newTaskFor() setzt voraus, dass der Pool nur PrioritizedTasks bekommt. Und submit() mit einem Callable läuft über die zweite Variante von newTaskFor(), die einen Callable statt eines Runnable entgegennimmt; sie müsstest du ebenso überschreiben.
Ist PriorityBlockingQueue threadsicher?
Ja, PriorityBlockingQueue ist threadsicher.
Die Threadsicherheit gewährleistet ein einzelnes ReentrantLock, das im Quelltext schlicht lock heißt. Es schützt alle öffentlichen Operationen, und an ihm hängt eine Condition namens notEmpty, auf die take() wartet, solange die Queue leer ist. Eine zweite Condition für eine volle Queue braucht die Klasse nicht – voll wird sie nie.
Eine Ausnahme macht das Vergrößern des Arrays. Ist es voll, gibt offer() das Lock frei und legt das größere Array unter einem einfachen Spinlock an: dem int-Feld allocationSpinLock, das per Compare-and-set von 0 auf 1 gesetzt wird. Danach nimmt offer() das Lock wieder und kopiert die Elemente um.
So können Consumer-Threads mit take() weiter entnehmen, während ein Producer-Thread das neue Array anlegt. Laut Kommentar im Quelltext verhindert das, dass wartende Consumer immer wieder zurückgestellt werden und sich Elemente aufstauen.
Das Array wächst unter 64 Plätzen auf etwas mehr als das Doppelte – aus 11 Plätzen werden 24, dann 50, dann 102 –, ab 64 Plätzen um die Hälfte.
Zusammenfassung und Ausblick
In diesem Artikel hast du die PriorityBlockingQueue kennengelernt. Sie ist threadsicher, blockierend und unbounded; ein einzelnes ReentrantLock schützt ihren Min-Heap. take() liefert immer das kleinste Element, der Iterator und toString() dagegen das Heap-Array in seiner internen Reihenfolge.
Als Work Queue eines ThreadPoolExecutors arbeitet sie Aufgaben nach Priorität ab – mit execute() direkt, mit submit() erst, wenn du newTaskFor() überschreibst.
Ab dem nächsten Teil der Tutorial-Serie stelle ich dir einige Queue-Implementierungen für Sonderfälle vor, beginnend mit der DelayQueue.
Hat dieser Artikel deine Fragen beantwortet? Dann freue ich mich über eine Bewertung auf meinem ProvenExpert-Profil – sie hilft anderen Entwickler:innen, diese Inhalte zu finden.
