Prioritätswarteschlange
Kurz gesagt: Eine Prioritätswarteschlange gibt immer das wichtigste Element zurück, zum Beispiel die dringendste Bestellung; auf einem Heap sind Einfügen und Entnehmen O(log n), und in Python ist es das Modul heapq.
Problem
Wer zuerst dran ist
Der Lieferdispatcher verteilt Bestellungen an die Kuriere. Jede Bestellung hat eine Frist: in wie vielen Minuten sie vor der Tür des Kunden sein muss. Ein freier Kurier bekommt immer die dringendste Bestellung.
Bestellungen kommen ständig herein. An einem Freitagabend in einer Großstadt warten 10.000 auf einmal, und 100.000 laufen an einem Tag durch.
Die erste Lösung ist eine nach Frist sortierte Liste. Bei jeder neuen Bestellung fügst du sie ein und sortierst die Liste neu. Sortieren kostet O(n log n) (siehe Mergesort), und das für jede Bestellung. Selbst die klügste Sortierung, die merkt, dass die Liste fast sortiert ist, muss sie mindestens durchlaufen: n Schritte bei jeder neuen Bestellung.
Und der Kurier braucht nicht die ganze Reihenfolge der Liste. Nur die erste.
Die Idee
Eine Warteschlange, die nur weiß, wer der Erste ist
Eine Struktur mit zwei Aktionen: eine Bestellung einfügen und die dringendste entnehmen. Das ist eine Prioritätswarteschlange. Sie verspricht keine Reihenfolge für alle, nur wer der Erste ist.
Genau das macht der Heap aus der Lektion zum Heapsort. Hier ist er umgedreht: ein Min-Heap, in dem jeder Elternknoten höchstens so groß ist wie seine Kinder. Dann steht die früheste Frist immer an Index 0.
- Entnehmen kennst du schon vom Heapsort. Die Spitze geht, das letzte Element nimmt ihren Platz ein und sinkt nach unten. Der Unterschied: Es sinkt zum dringenderen Kind.
- Einfügen ist das Neue: Aufsteigen. Die neue Bestellung kommt ans Ende des Arrays. Dann wird sie mit ihrem Elternknoten an Index (i - 1) / 2 verglichen. Ist sie dringender, tauschen sie die Plätze, und sie steigt weiter. Sie bleibt stehen, wenn der Elternknoten dringender ist oder wenn sie oben ankommt.
Sieh es dir an
Aufsteigen und Absinken
Einfügen: ans Ende, dann nach oben.
def push(self, order): self._items.append(order) i = len(self._items) - 1 while i > 0: parent = (i - 1) // 2 if not self._less(i, parent): return self._swap(i, parent) i = parentEntnehmen: Die Spitze geht, das letzte Element nimmt ihren Platz ein und sinkt.
def poll(self): items = self._items top = items[0] items[0] = items[-1] items.pop() if items: self._sift_down(0) return top
def _sift_down(self, i): n = len(self._items) while True: low = i left, right = 2 * i + 1, 2 * i + 2 if left < n and self._less(left, low): low = left if right < n and self._less(right, low): low = right if low == i: return self._swap(i, low) i = low__len__ ist die Methode, die len(queue) aufruft. So benutzt du die Warteschlange wie jede Liste.Sechs Bestellungen warten. Eine neue mit 18 Minuten kommt herein, und der Kurier nimmt die dringendste:
waiting = [ Order("A-101", 10), Order("A-102", 15), Order("A-103", 30), Order("A-104", 20), Order("A-105", 25), Order("A-106", 40),]queue = DeadlineQueue()for o in waiting: queue.push(o)print(f"Heap: {queue.deadlines()}")queue.push(Order("A-107", 18))print(f"Nach A-107: {queue.deadlines()}")order = queue.poll()print(f"Der Kurier nimmt {order.id} ({order.deadline} Min.)")print(f"Übrig: {queue.deadlines()}")Heap: [10, 15, 30, 20, 25, 40]Nach A-107: [10, 15, 18, 20, 25, 40, 30]Der Kurier nimmt A-101 (10 Min.)Übrig: [15, 20, 18, 30, 25, 40]class DeadlineQueue: def __init__(self): self._items = []
def push(self, order): self._items.append(order) i = len(self._items) - 1 while i > 0: parent = (i - 1) // 2 if not self._less(i, parent): return self._swap(i, parent) i = parent
def poll(self): items = self._items top = items[0] items[0] = items[-1] items.pop() if items: self._sift_down(0) return top
def _sift_down(self, i): n = len(self._items) while True: low = i left, right = 2 * i + 1, 2 * i + 2 if left < n and self._less(left, low): low = left if right < n and self._less(right, low): low = right if low == i: return self._swap(i, low) i = low
def _less(self, i, j): items = self._items return items[i].deadline < items[j].deadline
def _swap(self, i, j): items = self._items items[i], items[j] = items[j], items[i]
def __len__(self): return len(self._items)
def deadlines(self): return [o.deadline for o in self._items]Schritt für Schritt, in Worten
- Sechs wartende Bestellungen, die Fristen sind in Minuten. Ein Min-Heap: Jeder Elternknoten ist höchstens so groß wie seine Kinder, deshalb liegt die dringendste, 10, oben.
- push: Eine neue Bestellung mit 18 Minuten kommt ans Ende des Arrays, Index 6. Im Baum ist sie ein Kind von 30.
- Aufsteigen: Der Elternknoten von 6 ist (6 - 1) / 2 = 2. 18 ist dringender als 30, also tauschen sie.
- Der Elternknoten von 2 ist 0, mit 10 Minuten. 10 ist dringender als 18: Die Bestellung bleibt stehen. Ein Tausch, höchstens einer pro Ebene.
- poll: Der Kurier nimmt die Spitze, die Bestellung mit 10 Minuten. An Index 0 bleibt eine Lücke.
- Das letzte Element, 30, springt nach oben und das Array schrumpft auf sechs. Jetzt setzt _sift_down es an seinen Platz.
- Absinken, wie beim Heapsort, aber zum dringenderen Kind: 30 gegen 15 und 18 tauscht mit 15, dann gegen 20 und 25 tauscht mit 20.
- Index 3 hat keine Kinder: fertig. Oben liegt 15, die nächstdringende. Beide Bewegungen kosten O(log n): die Höhe des Baums.
Das Array ist nicht sortiert: 18 steht vor 20. Sortiert ist nur, was zählt: Die früheste Frist liegt oben.
Probier es aus
Sag die Ausgabe voraus
Fünf Bestellungen kommen in eine leere Warteschlange. Was steht danach im Array?
queue = DeadlineQueue()queue.push(Order("B-1", 40))queue.push(Order("B-2", 25))queue.push(Order("B-3", 30))queue.push(Order("B-4", 10))queue.push(Order("B-5", 35))print(queue.deadlines())Probier es aus
Ergänze
Beim Aufsteigen wird das Element an Index i mit seinem Elternknoten verglichen. Wo liegt der Elternknoten?
parent = Die Kinder von p liegen bei 2p + 1 und 2p + 2. Die Umkehrung ist (i - 1) / 2 mit ganzzahliger Division: 3 und 4 ergeben beide 1.
Die Idee
Was es kostet
- Einfügen: höchstens ein Vergleich pro Ebene. Ein Baum mit n Elementen hat etwa log₂ n Ebenen: O(log n).
- Entnehmen: zwei Vergleiche pro Ebene beim Absinken: ebenfalls O(log n).
- Wer der Erste ist: Die Spitze liegt an Index 0, ohne einen einzigen Vergleich: O(1).
Lass den Dispatcher mit beiden Versionen laufen. Die Bestellungen kommen nacheinander, nach jeder zweiten nimmt ein Kurier die dringendste, und am Ende nehmen die Kuriere den Rest. Gezählt werden die Vergleiche zwischen zwei Fristen:
- Sortierte Liste: nach jeder Bestellung neu sortieren
O(n²) - Min-Heap: nach oben und nach unten sieben
O(n log n)
| Algorithmus | n = 10 | n = 1.000 | n = 10.000 | n = 100.000 |
|---|---|---|---|---|
| Sortierte Liste: nach jeder Bestellung neu sortieren | 11 | 63.430 | 6.250.042 | ≈ 625.004.200 |
| Min-Heap: nach oben und nach unten sieben | 28 | 17.397 | 256.807 | 3.386.161 |
class DeadlineQueue: def __init__(self): self._items = []
def push(self, order): self._items.append(order) i = len(self._items) - 1 while i > 0: parent = (i - 1) // 2 if not self._less(i, parent): return self._swap(i, parent) i = parent
def poll(self): items = self._items top = items[0] items[0] = items[-1] items.pop() if items: self._sift_down(0) return top
def _sift_down(self, i): n = len(self._items) while True: low = i left, right = 2 * i + 1, 2 * i + 2 if left < n and self._less(left, low): low = left if right < n and self._less(right, low): low = right if low == i: return self._swap(i, low) i = low
def _less(self, i, j): items = self._items return items[i].deadline < items[j].deadline
def _swap(self, i, j): items = self._items items[i], items[j] = items[j], items[i]
def __len__(self): return len(self._items)
def deadlines(self): return [o.deadline for o in self._items]Bei 10 Bestellungen macht die sortierte Liste 11 Vergleiche und der Heap 28: Bei wenigen Bestellungen ist die Liste völlig in Ordnung. Bei 10.000 sind es 6.250.042 gegen 256.807. Bei 100.000 macht der Heap 3.386.161, die Liste etwa 625 Millionen. Das lässt Bench nicht laufen, es schätzt es aus der Kurve.
Die sortierte Liste in Bench ist im besten Fall: Die neue Bestellung rutscht nur an ihren Platz, nichts anderes wird sortiert. Es bleibt trotzdem O(n) pro Bestellung, weil sie im Schnitt die halbe Liste durchläuft. Der Heap ist O(log n) pro Bestellung.
Genauer, für Neugierige
Welche Sortierung merkt, dass die Liste fast sortiert ist? TimSort, das sort in Python, List.sort und Collections.sort in Java und Array.prototype.sort in JavaScript verwenden (siehe die eingebaute Sortierung). Es findet den sortierten Teil mit etwa n Vergleichen und setzt das neue Element dort ein: O(n) pro Bestellung. Bench zählt eine noch billigere Variante, das Rutschen vom Ende her, und es kommt trotzdem O(n) heraus. C# List.Sort ist Introsort und sucht keine sortierten Teile: Bei jeder Bestellung macht es etwa n log n Vergleiche, noch mehr.
Warum ist das Aufsteigen auch in der Praxis schnell? Eine neue Bestellung liegt meist irgendwo in der Mitte der Dringlichkeit. Die Hälfte der Elemente eines Heaps sind Blätter, ein Viertel liegt direkt darüber, und so weiter. Im Schnitt bleibt eine zufällige Bestellung also nach ein, zwei Ebenen stehen. Der schlimmste Fall (die neue ist die dringendste von allen) sind trotzdem nur log₂ n.
Probier es aus
Finde den Fehler
Ein Kollege hat das Absinken geschrieben, aber die Kuriere fahren manchmal mit einer späteren Bestellung vor einer früheren los:
def _sift_down(self, i): n = len(self._items) while True: low = i left, right = 2 * i + 1, 2 * i + 2 if left < n and self._less(left, low): low = left if right < n and self._less(right, i): low = right if low == i: return self._swap(i, low) i = lowqueue = DeadlineQueue()queue.push(Order("C-1", 5))queue.push(Order("C-2", 10))queue.push(Order("C-3", 20))queue.push(Order("C-4", 30))queue.push(Order("C-5", 40))queue.push(Order("C-6", 50))taken = []while len(queue) > 0: taken.append(queue.poll().deadline)print(taken)[5, 20, 10, 30, 40, 50]In welcher Zeile steckt der Fehler?
Wähle die Zeile mit dem Fehler.
Das rechte Kind muss mit dem bisher dringendsten (low) verglichen werden, nicht mit dem Elternknoten (i). Hier sind beide Kinder dringender als 50, also bestehen beide Prüfungen, und das rechte, 20, gewinnt, obwohl das linke, 10, dringender ist. 20 landet über 10, und der Heap ist kaputt. Mit low lautet die Ausgabe [5, 10, 20, 30, 40, 50].
Sieh es dir an
Die fertigen Warteschlangen jeder Sprache
In einem Projekt schreibst du keinen Heap selbst. Fast jede Sprache hat eine Prioritätswarteschlange, und jede hat ihre Eigenheit.
In Python ist heapq ein Min-Heap auf einer gewöhnlichen Liste: heappush lässt aufsteigen, heappop nimmt die Spitze, und queue[0] ist die Spitze. Eine Bestellung geht als Tupel (deadline, id) hinein: Tupel werden von links nach rechts verglichen, also entscheidet zuerst die Frist.
Die Eigenheit: Die Liste ist ein Heap, keine sortierte Liste. Nur heappop entnimmt sie der Reihe nach.
import heapq
queue = []heapq.heappush(queue, (30, "A-101"))heapq.heappush(queue, (10, "A-102"))heapq.heappush(queue, (20, "A-103"))heapq.heappush(queue, (25, "A-104"))print(f"Am dringendsten: {queue[0][1]}")print("Innen:", *[deadline for deadline, _ in queue])while queue: deadline, order_id = heapq.heappop(queue) print(f"{order_id} ({deadline} Min.)")Am dringendsten: A-102Innen: 10 25 20 30A-102 (10 Min.)A-103 (20 Min.)A-104 (25 Min.)A-101 (30 Min.)Die zweite Eigenheit ist die Gleichheit. Bei gleichen Fristen vergleicht Python das Nächste im Tupel. Steht dort ein dict oder ein Objekt ohne <, schlägt die Warteschlange fehl:
import heapq
queue = []heapq.heappush(queue, (20, {"id": "A-101"}))try: heapq.heappush(queue, (20, {"id": "A-102"}))except TypeError as e: print("TypeError:", e)TypeError: '<' not supported between instances of 'dict' and 'dict'Die Lösung ist eine laufende Nummer zwischen Frist und Bestellung. Die Nummern sind nie gleich, also wird das dict nie verglichen, und bei gleichen Fristen kommt die frühere Bestellung zuerst:
import heapqfrom itertools import count
queue = []seq = count()heapq.heappush(queue, (20, next(seq), {"id": "A-101"}))heapq.heappush(queue, (20, next(seq), {"id": "A-102"}))heapq.heappush(queue, (10, next(seq), {"id": "A-103"}))while queue: deadline, _, order = heapq.heappop(queue) print(order["id"], deadline)A-103 10A-101 20A-102 20Genauer, für Neugierige
Keine dieser Warteschlangen ist stabil: Zwei Bestellungen mit derselben Frist können in beliebiger Reihenfolge herauskommen (siehe Stabilität). Wenn die Reihenfolge des Eintreffens wichtig ist, füge eine laufende Nummer hinzu, wie oben in Python.
Die Frist einer Bestellung, die schon wartet, lässt sich in keiner der drei ändern. remove in Java sucht das Element linear: O(n). Die übliche Lösung ist, die Bestellung mit der neuen Frist noch einmal einzufügen und die alte Kopie zu überspringen, wenn sie herauskommt. So arbeitet auch Dijkstra im kommenden Modul über Graphen (algo-graphs).
Wie es in den anderen Sprachen ist: Java hat PriorityQueue, C# PriorityQueue<TElement, TPriority> (seit .NET 6), Python heapq (Funktionen auf einer Liste), und JavaScript hat keine eingebaute Prioritätswarteschlange.
Sieh es dir an
Zwei weitere Aufgaben für einen Heap
Die obersten k. Welche 10 von einer Million Produkten verkaufen sich am besten? Ein Min-Heap, der nie mehr als 10 Elemente enthält: Jedes neue wird mit der Spitze verglichen. Das steht in der Lektion zum Heap, mit einem Bench dazu.
k sortierte Listen zusammenführen. Drei Lager schicken ihre Bestellungen nach Frist sortiert. Du brauchst eine gemeinsame Route, wieder nach Frist. Der Heap enthält nur den Kopf jeder Liste. Du nimmst den frühesten, und an seine Stelle setzt du den nächsten aus demselben Lager:
import heapq
lists = [[5, 20, 45], [10, 15, 50], [12, 30]]names = ["Nord", "Süd", "Mitte"]heads = [(lst[0], k, 0) for k, lst in enumerate(lists)]heapq.heapify(heads)route = []while heads: deadline, k, i = heapq.heappop(heads) route.append(f"{deadline} {names[k]}") if i + 1 < len(lists[k]): heapq.heappush(heads, (lists[k][i + 1], k, i + 1))print("Route:", ", ".join(route))Route: 5 Nord, 10 Süd, 12 Mitte, 15 Süd, 20 Nord, 30 Mitte, 45 Nord, 50 SüdDer Heap enthält nie mehr als k Elemente (hier 3). N Bestellungen kosten also O(N log k), nicht O(N log N) wie beim Sortieren von allem. So werden auch die sortierten Teile einer Datei zusammengeführt, die nicht in den Speicher passt.
Genauer, für Neugierige
heapq.merge(*lists) macht genau dasselbe und liefert die Zusammenführung Element für Element, ohne sie in einer Liste zu sammeln.Mergesort führt auch sortierte Teile zusammen, aber zwei auf einmal (Mergesort). Ein Heap führt k auf einmal zusammen.
Übung
Eine faire Warteschlange
Löse die Aufgabe in deinem Editor.
Gleiche Frist, die frühere zuerst. Baue eine Warteschlange FairQueue, in der bei gleicher Frist die Bestellung zuerst herauskommt, die früher gekommen ist. Speichere zu jeder Bestellung eine laufende Nummer und vergleiche zuerst nach Frist, dann nach Nummer. Lass sie mit den Bestellungen A-1 (30), A-2 (20), A-3 (30), A-4 (20), A-5 (10), A-6 (20) in dieser Reihenfolge laufen:
Reihenfolge: A-5, A-2, A-4, A-6, A-1, A-3Weitere Aufgaben
- Max-Heap. Dreh den Vergleich um und baue eine Warteschlange, die zuerst die teuerste Bestellung liefert.
- Vergleiche zählen. Füge dem Aufsteigen und dem Absinken einen Zähler hinzu und lass 1000 zufällige Bestellungen laufen. Wie viele Vergleiche fallen auf eine Bestellung?
- Die obersten drei. Wie bekommst du die drei dringendsten Bestellungen, ohne die anderen aus der Warteschlange zu nehmen?
Test