Python heapq: Prioritätswarteschlangen und Heap-Operationen leicht gemacht

Name

Soren Atelier

Updated on 10.2.2026

Sie müssen Elemente nach Priorität verarbeiten -- Aufgaben in einem Job-Scheduler, Ereignisse in einer Simulation oder den kürzesten Pfad in einem Graphen. Eine sortierte Liste scheint die offensichtliche Wahl zu sein, aber das Einfügen in eine sortierte Liste ist O(n), was bei der Verarbeitung von Millionen von Elementen zum Engpass wird. Die gesamte Liste bei jedem neuen Element neu zu sortieren ist mit O(n log n) noch schlimmer.

Die Kosten summieren sich schnell. Ein Job-Scheduler, der seine Warteschlange bei jedem Einfügen neu sortiert, verschwendet den Großteil seiner Zeit mit Verwaltungsaufwand. Ein Pfadfindungsalgorithmus, der bei jedem Schritt eine vollständige Liste nach dem Minimum durchsucht, macht aus einer linearen Operation eine quadratische.

Pythons heapq-Modul bietet eine Heap-basierte Prioritätswarteschlange mit O(log n) für das Einfügen und O(log n) für das Extrahieren des kleinsten Elements. Es enthält auch optimierte Funktionen zum Finden der N größten oder kleinsten Elemente aus einem beliebigen Iterable, ohne die gesamte Sammlung zu sortieren. Diese Anleitung behandelt jede Funktion in heapq mit praktischen Beispielen.

Was ist ein Heap?

Ein Heap ist ein Binärbaum, bei dem jeder Elternknoten kleiner oder gleich seinen Kindern ist (Min-Heap). Pythons heapq implementiert einen Min-Heap unter Verwendung einer regulären Liste, wobei das kleinste Element immer an Index 0 steht.

Die wichtigsten Operationen und ihre Zeitkomplexitäten:

Grundlegende Heap-Operationen

Erstellen und Verwenden eines Heaps

import heapq
 
# Start with an empty heap (just a regular list)
heap = []
 
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
 
print(heap)        # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0])     # 1 (smallest element, always at index 0)
 
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest)    # 1
print(heap)        # [2, 5, 3]

Eine Liste in einen Heap umwandeln

import heapq
 
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data)  # In-place, O(n)
print(data)  # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
 
# Pop elements in sorted order
sorted_data = []
while data:
    sorted_data.append(heapq.heappop(data))
print(sorted_data)  # [1, 2, 3, 4, 5, 6, 7, 8]

heappushpop und heapreplace

Diese kombinierten Operationen sind effizienter als separate push/pop-Aufrufe:

import heapq
 
heap = [1, 3, 5, 7]
heapq.heapify(heap)
 
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result)  # 1 (pushed 2, popped 1)
print(heap)    # [2, 3, 5, 7]
 
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result)  # 2 (popped 2, pushed 10)
print(heap)    # [3, 7, 5, 10]

nlargest und nsmallest

Diese Funktionen finden die N größten oder kleinsten Elemente, ohne die gesamte Sammlung zu sortieren.

import heapq
 
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
 
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3)  # [97, 95, 92]
 
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3)  # [69, 72, 78]

Mit einer Schlüsselfunktion

Beide akzeptieren einen key-Parameter für benutzerdefinierte Vergleiche:

import heapq
 
students = [
    {'name': 'Alice', 'gpa': 3.9},
    {'name': 'Bob', 'gpa': 3.5},
    {'name': 'Charlie', 'gpa': 3.8},
    {'name': 'Diana', 'gpa': 3.95},
    {'name': 'Eve', 'gpa': 3.2},
]
 
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
    print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8

Wann nlargest/nsmallest vs sorted() verwenden

Implementierung eines Max-Heaps

Pythons heapq bietet nur einen Min-Heap. Um einen Max-Heap zu erstellen, negieren Sie die Werte:

import heapq
 
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
    heapq.heappush(max_heap, -val)
 
# Pop largest items
while max_heap:
    print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1

Prioritätswarteschlange mit benutzerdefinierten Objekten

Für Objekte, die nicht direkt vergleichbar sind, verwenden Sie Tupel (priority, counter, item):

import heapq
from dataclasses import dataclass
 
@dataclass
class Task:
    name: str
    description: str
 
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
 
def add_task(priority, task):
    global counter
    heapq.heappush(task_queue, (priority, counter, task))
    counter += 1
 
def get_next_task():
    if task_queue:
        priority, _, task = heapq.heappop(task_queue)
        return task
    return None
 
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
 
while task_queue:
    task = get_next_task()
    print(f"{task.name}: {task.description}")
 
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logs

Der Zähler dient als Gleichstandsbrecher, damit Elemente mit gleicher Priorität in Einfügereihenfolge (FIFO) zurückgegeben werden.

Praktische Anwendungsfälle

Sortierte Iterables zusammenführen

import heapq
 
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
 
merged = list(heapq.merge(list1, list2, list3))
print(merged)  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

heapq.merge() ist speichereffizient -- es lädt nicht alle Listen gleichzeitig in den Speicher.

Laufender Median

import heapq
 
class RunningMedian:
    """Track the median of a stream of numbers."""
    def __init__(self):
        self.low = []   # max-heap (negated values)
        self.high = []  # min-heap
 
    def add(self, num):
        heapq.heappush(self.low, -num)
        heapq.heappush(self.high, -heapq.heappop(self.low))
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))
 
    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2
 
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
    rm.add(num)
    print(f"Added {num}, median = {rm.median()}")

Dijkstras kürzester Pfad

import heapq
 
def dijkstra(graph, start):
    """Find shortest paths from start to all nodes."""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]
 
    while pq:
        dist, node = heapq.heappop(pq)
        if dist > distances[node]:
            continue
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))
 
    return distances
 
graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
 
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Heap-Leistung in Jupyter analysieren

Beim Vergleich von Heap-basierten Algorithmen mit Alternativen oder beim Profiling der Leistung von Prioritätswarteschlangen beschleunigt interaktives Experimentieren den Prozess. RunCell (opens in a new tab) ist ein KI-Agent für Jupyter, der Ihnen helfen kann, Heap-Operationen zu benchmarken, Leistungscharakteristiken zu visualisieren und Ihre Datenstruktur-Entscheidungen interaktiv zu optimieren.

FAQ

Was ist heapq in Python?

heapq ist ein Standardbibliotheksmodul, das einen Min-Heap unter Verwendung einer regulären Python-Liste implementiert. Es bietet Funktionen zum Einfügen und Entfernen von Elementen in O(log n) Zeit, während die Heap-Eigenschaft beibehalten wird (kleinstes Element immer an Index 0). Es ist der Standardweg zur Implementierung von Prioritätswarteschlangen in Python.

Wie erstelle ich einen Max-Heap in Python?

Pythons heapq unterstützt nur Min-Heaps. Um einen Max-Heap zu simulieren, negieren Sie die Werte beim Einfügen und erneut beim Entfernen: heapq.heappush(heap, -value) zum Einfügen, -heapq.heappop(heap) zum Abrufen des größten Elements. Für komplexe Objekte negieren Sie die Priorität in einem Tupel: (-priority, counter, item).

Wann sollte ich heapq vs sorted() verwenden?

Verwenden Sie heapq.nlargest(n, data) oder heapq.nsmallest(n, data), wenn Sie nur die Top-N-Elemente benötigen und N im Verhältnis zur Gesamtdatengröße klein ist. Verwenden Sie sorted(), wenn Sie alle Elemente in der Reihenfolge benötigen oder N nahe der Gesamtgröße liegt. Verwenden Sie min()/max() wenn N gleich 1 ist.

Ist heapq threadsicher?

Nein. Einzelne heappush- und heappop-Aufrufe sind nicht atomar. Wenn mehrere Threads auf denselben Heap zugreifen, benötigen Sie externe Synchronisation. Für threadsichere Prioritätswarteschlangen verwenden Sie queue.PriorityQueue, das heapq mit ordnungsgemäßer Sperrung umhüllt.

Was ist die Zeitkomplexität der heapq-Operationen?

heappush und heappop sind beide O(log n). heapify wandelt eine Liste in O(n) in einen Heap um. nlargest(k, data) und nsmallest(k, data) laufen in O(n log k), wobei n die Datengröße und k die Anzahl der angeforderten Elemente ist.

Fazit

Pythons heapq-Modul ist das Standardwerkzeug für Prioritätswarteschlangen und Top-N-Auswahl. Sein Heap-basierter Ansatz bietet O(log n) Push- und Pop-Operationen, was es dramatisch schneller macht als die Pflege einer sortierten Liste. Verwenden Sie heappush/heappop für Prioritätswarteschlangen-Muster, nlargest/nsmallest für Top-N-Auswahl, merge zum Kombinieren sortierter Sequenzen und heapify, wenn Sie eine bestehende Liste umwandeln möchten.