Master priority queues — what they are, how they differ from regular queues, heap-based implementation, Python
Introduction
A priority queue is a special queue where each element has an associated priority. Elements are dequeued in order of their priority — not in the order they were inserted.
In a regular queue: first in, first out. In a priority queue: highest priority element is always dequeued first.
| Regular Queue | [Job1, Job2, Job3] (FIFO) |
| Dequeue | Job1 |
| Priority Queue | [(Job3, priority=10), (Job1, priority=3), (Job2, priority=7)] |
| Dequeue | Job3 (priority 10, highest) |
| Dequeue | Job2 (priority 7) |
| Dequeue | Job1 (priority 3, lowest) |
Operations and Complexity
| Operation | Description | Time |
|---|
insert(item, priority) | Add element with priority | O(log n) |
extract_min() or extract_max() | Remove and return highest-priority element | O(log n) |
peek_min() or peek_max() | View highest-priority without removing | O(1) |
is_empty() | True if empty | O(1) |
size() | Number of elements | O(1) |
decrease_key() | Update priority (used in Dijkstra) | O(log n) |
Python's heapq Module
Python provides a min-heap in heapq:
import heapq
# Min-heap
heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 7)
heapq.heappush(heap, 2)
print(heap) # [1, 2, 3, 7, 5] — heap-ordered (not fully sorted)
print(heap[0]) # 1 — peek minimum: O(1)
print(heapq.heappop(heap)) # 1 — extract minimum: O(log n)
print(heapq.heappop(heap)) # 2
# Build heap from list: O(n) (better than pushing n items: O(n log n))
nums = [5, 1, 3, 7, 2]
heapq.heapify(nums) # In-place heapify: O(n)
# Max-heap workaround: negate values
max_heap = []
heapq.heappush(max_heap, -5)
heapq.heappush(max_heap, -1)
heapq.heappush(max_heap, -8)
max_val = -heapq.heappop(max_heap) # -(-8) = 8 — maximum
# Priority queue with (priority, item) tuples:
pq = []
heapq.heappush(pq, (1, "low priority task"))
heapq.heappush(pq, (10, "critical task"))
heapq.heappush(pq, (5, "medium task"))
priority, task = heapq.heappop(pq)
print(f"Processing: {task} (priority={priority})") # critical task (priority=10)? NO!
# heapq is MIN-heap, so priority=1 comes first.
# For max-priority-queue behavior, use negative priorities:
pq2 = []
heapq.heappush(pq2, (-10, "critical task"))
heapq.heappush(pq2, (-5, "medium task"))
heapq.heappush(pq2, (-1, "low task"))
neg_p, task = heapq.heappop(pq2)
print(f"Processing: {task} (priority={-neg_p})") # critical task (priority=10)
Complete Priority Queue Class
import heapq
from typing import Any
class MinPriorityQueue:
"""
Min-priority queue: lowest priority value = highest urgency.
Wraps Python's heapq with a clean interface.
"""
def __init__(self):
self._heap = [] # List of (priority, counter, item)
self._counter = 0 # Tiebreaker for equal priorities (FIFO among equals)
def insert(self, item: Any, priority: float) -> None:
"""Insert item with given priority. O(log n)."""
heapq.heappush(self._heap, (priority, self._counter, item))
self._counter += 1
def extract_min(self) -> Any:
"""Remove and return item with lowest priority value. O(log n)."""
if self.is_empty():
raise IndexError("Priority queue is empty")
priority, _, item = heapq.heappop(self._heap)
return item
def peek_min(self) -> Any:
"""View minimum-priority item without removing. O(1)."""
if self.is_empty():
raise IndexError("Priority queue is empty")
return self._heap[0][2]
def min_priority(self) -> float:
"""Get the minimum priority value. O(1)."""
if self.is_empty():
raise IndexError("Priority queue is empty")
return self._heap[0][0]
def is_empty(self) -> bool:
return len(self._heap) == 0
def size(self) -> int:
return len(self._heap)
def __len__(self) -> int:
return len(self._heap)
def __repr__(self) -> str:
return f"MinPQ({[(p, item) for p, _, item in self._heap]})"
Application 1 — K Largest Elements
def k_largest_elements(nums, k):
"""
Find the k largest elements using a min-heap of size k.
Time: O(n log k), Space: O(k)
Much better than O(n log n) for sorting when k << n.
"""
heap = [] # Min-heap of size k
for num in nums:
heapq.heappush(heap, num)
if len(heap) > k:
heapq.heappop(heap) # Remove smallest to maintain k largest
return sorted(heap, reverse=True)
print(k_largest_elements([3,1,4,1,5,9,2,6,5,3], 4)) # [9, 6, 5, 5]
Alternative — k-th largest element:
def kth_largest(nums, k):
"""Find k-th largest element. O(n log k)."""
heap = []
for num in nums:
heapq.heappush(heap, num)
if len(heap) > k:
heapq.heappop(heap)
return heap[0] # Top of min-heap of size k = k-th largest
print(kth_largest([3,2,1,5,6,4], 2)) # 5 (2nd largest)
Application 2 — Merge K Sorted Lists
def merge_k_sorted(lists):
"""
Merge k sorted lists. O(n log k) where n = total elements.
"""
heap = [] # (value, list_index, element_index)
for i, lst in enumerate(lists):
if lst:
heapq.heappush(heap, (lst[0], i, 0))
result = []
while heap:
val, list_idx, elem_idx = heapq.heappop(heap)
result.append(val)
if elem_idx + 1 < len(lists[list_idx]):
next_val = lists[list_idx][elem_idx + 1]
heapq.heappush(heap, (next_val, list_idx, elem_idx + 1))
return result
print(merge_k_sorted([[1,4,7],[2,5,8],[3,6,9]])) # [1,2,3,4,5,6,7,8,9]
Application 3 — Task Scheduler (CPU Scheduling)
import heapq
from collections import Counter
def least_interval(tasks, n):
"""
CPU Scheduler: Given tasks and cooldown n between same tasks,
find minimum time to complete all tasks.
Time: O(T log T) where T = unique task count.
"""
freq = Counter(tasks)
max_heap = [-f for f in freq.values()] # Max-heap (negate for max)
heapq.heapify(max_heap)
time = 0
while max_heap:
cycle = []
for _ in range(n + 1): # Process n+1 tasks per cycle
if max_heap:
cycle.append(-heapq.heappop(max_heap)) # Get most frequent
# Decrement frequencies
for f in cycle:
if f - 1 > 0:
heapq.heappush(max_heap, -(f - 1))
time += len(cycle) if not max_heap else n + 1
return time
print(least_interval(["A","A","A","B","B","B"], 2)) # 8
# A→B→_→A→B→_→A→B (where _ = idle)
Application 4 — Dijkstra's Shortest Path
Priority queue is the heart of Dijkstra's algorithm:
import heapq
def dijkstra(graph, source):
"""
Dijkstra's shortest path algorithm.
graph: dict {node: [(neighbor, weight), ...]}
Time: O((V + E) log V)
"""
distances = {node: float('inf') for node in graph}
distances[source] = 0
# (distance, node)
pq = [(0, source)]
while pq:
dist, node = heapq.heappop(pq)
if dist > distances[node]:
continue # Outdated entry
for neighbor, weight in graph.get(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}
Python's queue.PriorityQueue (Thread-Safe)
from queue import PriorityQueue
pq = PriorityQueue()
pq.put((3, "medium")) # (priority, item)
pq.put((1, "high"))
pq.put((5, "low"))
while not pq.empty():
priority, item = pq.get()
print(f"{priority}: {item}")
# 1: high
# 3: medium
# 5: low
Summary
| Feature | Priority Queue |
|---|
| Order | By priority, not insertion order |
| Insert | O(log n) |
| Extract min/max | O(log n) |
| Peek | O(1) |
| Implementation | Binary heap (most common) |
| Python | heapq module (min-heap) |
| Thread-safe | queue.PriorityQueue |
Key applications:
- K largest/smallest elements: Min-heap of size k
- Merge k sorted lists: Always extract the current minimum across all lists
- Dijkstra's algorithm: Always process the unvisited node with smallest distance
- Task scheduling: Always process highest-priority task
- Huffman encoding: Always merge two lowest-frequency trees
*Next Lesson: Queue Applications — BFS, level-order traversal, 01-BFS, and more.*