Master priority queues built on heaps — complete Python and Java implementations, Dijkstra
Priority Queue vs Regular Queue
Regular queue: First In, First Out — always dequeues oldest element. Priority queue: Always dequeues highest priority element, regardless of insertion order.
Regular Queue: [task1, task2, task3] → dequeue task1 (oldest)
Priority Queue: [(task2, priority=10), (task1, priority=3), (task3, priority=7)]
→ dequeue task2 (highest priority)
Binary heaps provide all priority queue operations in O(log n).
Application 1 — Dijkstra's Shortest Path
import heapq
from collections import defaultdict
def dijkstra(graph, source):
"""
Dijkstra's algorithm using priority queue (min-heap).
graph: {node: [(neighbor, weight), ...]}
Time: O((V + E) log V)
"""
dist = defaultdict(lambda: float('inf'))
dist[source] = 0
# Min-heap: (distance, node)
pq = [(0, source)]
while pq:
d, u = heapq.heappop(pq)
if d > dist[u]:
continue # Outdated entry (lazy deletion)
for v, w in graph.get(u, []):
new_dist = d + w
if new_dist < dist[v]:
dist[v] = new_dist
heapq.heappush(pq, (new_dist, v))
return dict(dist)
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}
import heapq
class MedianFinder:
"""
Find median of stream using two heaps.
- Max-heap: lower half of data
- Min-heap: upper half of data
Balance: lower half has same or one more element.
Median = top of lower (odd total) or average of both tops (even).
Insert: O(log n), FindMedian: O(1)
"""
def __init__(self):
self.lower = [] # Max-heap (negate for Python's min-heap)
self.upper = [] # Min-heap
def add_num(self, num):
# Add to lower half first (max-heap)
heapq.heappush(self.lower, -num)
# Balance: if lower top > upper bottom, move to upper
if self.upper and -self.lower[0] > self.upper[0]:
heapq.heappush(self.upper, -heapq.heappop(self.lower))
# Balance sizes: lower has at most 1 more than upper
if len(self.lower) > len(self.upper) + 1:
heapq.heappush(self.upper, -heapq.heappop(self.lower))
elif len(self.upper) > len(self.lower):
heapq.heappush(self.lower, -heapq.heappop(self.upper))
def find_median(self):
if len(self.lower) > len(self.upper):
return -self.lower[0] # Odd total: lower has extra
return (-self.lower[0] + self.upper[0]) / 2 # Even total: average
mf = MedianFinder()
for num in [1, 2, 3, 4, 5]:
mf.add_num(num)
print(f"After adding {num}: median = {mf.find_median()}")
# 1, 1.5, 2, 2.5, 3
Application 3 — Task Scheduler (CPU)
import heapq
from collections import Counter
def task_scheduler(tasks, n):
"""
Minimum intervals for CPU with cooldown n between same tasks.
Always execute the most frequent remaining task.
"""
freq = Counter(tasks)
max_heap = [-f for f in freq.values()]
heapq.heapify(max_heap)
time = 0
cooldown_queue = [] # (available_time, neg_freq)
while max_heap or cooldown_queue:
time += 1
# Release tasks that are now available
if cooldown_queue and cooldown_queue[0][0] <= time:
_, f = heapq.heappop(cooldown_queue)
heapq.heappush(max_heap, f)
if max_heap:
freq_left = heapq.heappop(max_heap) + 1 # Decrement (negated)
if freq_left < 0: # Still has remaining instances
heapq.heappush(cooldown_queue, (time + n + 1, freq_left))
# else: idle
return time
print(task_scheduler(["A","A","A","B","B","B"], 2)) # 8
# A→B→idle→A→B→idle→A→B
Application 4 — K-Way Merge
import heapq
def k_way_merge(sorted_lists):
"""
Merge k sorted lists into one sorted list.
Time: O(n log k), Space: O(k)
"""
result = []
heap = [] # (value, list_idx, element_idx)
for i, lst in enumerate(sorted_lists):
if lst:
heapq.heappush(heap, (lst[0], i, 0))
while heap:
val, li, ei = heapq.heappop(heap)
result.append(val)
if ei + 1 < len(sorted_lists[li]):
heapq.heappush(heap, (sorted_lists[li][ei+1], li, ei+1))
return result
print(k_way_merge([[1,4,7],[2,5,8],[3,6,9]]))
# [1, 2, 3, 4, 5, 6, 7, 8, 9]
Application 5 — Event Simulation
import heapq
def simulate_events(events):
"""
Process events in time order using priority queue.
events: list of (time, event_type, data)
"""
event_queue = list(events)
heapq.heapify(event_queue)
results = []
while event_queue:
time, event_type, data = heapq.heappop(event_queue)
print(f"t={time}: {event_type} — {data}")
results.append((time, event_type))
return results
events = [
(5, "SERVER_REQUEST", "req_3"),
(1, "SERVER_START", "server"),
(3, "CLIENT_CONNECT", "client_1"),
(2, "DATABASE_READY", "db"),
]
simulate_events(events)
# t=1: SERVER_START
# t=2: DATABASE_READY
# t=3: CLIENT_CONNECT
# t=5: SERVER_REQUEST
Java's PriorityQueue
import java.util.PriorityQueue;
// Min-heap (natural ordering)
PriorityQueue<Integer> pq = new PriorityQueue<>();
pq.offer(5); pq.offer(1); pq.offer(3);
System.out.println(pq.poll()); // 1 (minimum)
// Max-heap (reverse ordering)
PriorityQueue<Integer> maxPQ = new PriorityQueue<>(Collections.reverseOrder());
maxPQ.offer(5); maxPQ.offer(1); maxPQ.offer(3);
System.out.println(maxPQ.poll()); // 5 (maximum)
// Custom priority: sort by string length
PriorityQueue<String> strPQ = new PriorityQueue<>(Comparator.comparingInt(String::length));
strPQ.offer("banana"); strPQ.offer("cat"); strPQ.offer("ab");
System.out.println(strPQ.poll()); // "ab" (shortest)
Summary
Priority queues power a huge range of algorithms:
| Application | Heap Type | Pattern |
|---|
| Dijkstra's | Min-heap | Always process smallest distance |
| Prim's MST | Min-heap | Always add cheapest edge |
| Huffman Coding | Min-heap | Always merge two smallest frequencies |
| Median Stream | Two heaps | Balance upper and lower halves |
| Top-K | Min-heap of size k | Maintain k largest |
| K-Way Merge | Min-heap | Always take smallest front |
| Task Scheduler | Max-heap | Always execute most frequent |
| Event Simulation | Min-heap | Process earliest event first |
The heap-based priority queue is one of the most versatile data structures — whenever you need to repeatedly identify and process the extreme (minimum or maximum) of a changing collection, a heap is the right tool.
*Next Lesson: Heap Problems*