- Compare implementations: heap O(log n) insert/extract, unsorted array O(1)/O(n), sorted array O(n)/O(1)
- Understand binary heap is best all-around choice for priority queues
- Build custom PriorityQueue class with heapq, counter, and max/min support
Priority Queue Implementations
Priority queues are an interface. Multiple ways to implement them. Each with trade-offs.
Binary Heap (Most Common)
The standard. Balanced performance.
Structure: Complete binary tree stored in array.
Operations:
- Insert: O(log n) - add at end, bubble up
- Extract: O(log n) - move last to root, bubble down
- Peek: O(1) - just return root
import heapq
class MinPriorityQueue:
def __init__(self):
self.heap = []
def insert(self, priority, item):
heapq.heappush(self.heap, (priority, item))
def extract_min(self):
if not self.heap:
raise IndexError("Empty queue")
return heapq.heappop(self.heap)
def peek(self):
if not self.heap:
return None
return self.heap[0]
def is_empty(self):
return len(self.heap) == 0
def size(self):
return len(self.heap)
Why binary heap? Good balance. O(log n) for insert and extract. O(1) peek. Compact array storage.
Unsorted Array
Simplest implementation.
Insert: O(1) - just append to array.
Extract: O(n) - scan entire array for max/min, remove.
class UnsortedArrayPQ:
def __init__(self):
self.items = []
def insert(self, priority, item):
self.items.append((priority, item)) # O(1)
def extract_max(self):
if not self.items:
raise IndexError("Empty queue")
# Find max - O(n)
max_idx = 0
for i in range(1, len(self.items)):
if self.items[i][0] > self.items[max_idx][0]:
max_idx = i
return self.items.pop(max_idx)
When to use? Rare extracts, frequent inserts. But usually heap is better.
Sorted Array
Keep array sorted by priority.
Insert: O(n) - find position with binary search (O(log n)), shift elements (O(n)).
Extract: O(1) - just pop from end (max) or start (min).
class SortedArrayPQ:
def __init__(self):
self.items = []
def insert(self, priority, item):
# Binary search for position
left, right = 0, len(self.items)
while left < right:
mid = (left + right) // 2
if self.items[mid][0] < priority:
left = mid + 1
else:
right = mid
self.items.insert(left, (priority, item)) # O(n) shift
def extract_max(self):
if not self.items:
raise IndexError("Empty queue")
return self.items.pop() # O(1)
When to use? Rare inserts, frequent extracts. Still, heap usually better.
Binary Search Tree (BST)
Use BST (or balanced BST like AVL).
Insert: O(log n) balanced, O(n) worst case unbalanced.
Extract: O(log n) balanced - find max/min, delete.
# Conceptual - use library for production
class BSTNode:
def __init__(self, priority, item):
self.priority = priority
self.item = item
self.left = None
self.right = None
class BSTPQ:
def __init__(self):
self.root = None
def insert(self, priority, item):
self.root = self._insert(self.root, priority, item)
def _insert(self, node, priority, item):
if node is None:
return BSTNode(priority, item)
if priority < node.priority:
node.left = self._insert(node.left, priority, item)
else:
node.right = self._insert(node.right, priority, item)
return node
def extract_max(self):
if self.root is None:
raise IndexError("Empty queue")
# Find and remove rightmost (max)
# Implementation omitted for brevity
pass
Heap is simpler. BST useful if you need range queries or iteration in order.
Comparison
| Implementation | Insert | Extract | Peek | Space | Notes |
|---|---|---|---|---|---|
| Binary Heap | O(log n) | O(log n) | O(1) | O(n) | Best all-around |
| Unsorted Array | O(1) | O(n) | O(n) | O(n) | Only if rare extracts |
| Sorted Array | O(n) | O(1) | O(1) | O(n) | Only if rare inserts |
| BST | O(log n)* | O(log n)* | O(log n)* | O(n) | Needs balancing |
| Fibonacci Heap | O(1)** | O(log n)** | O(1) | O(n) | Complex, rarely used |
*Balanced BST, otherwise O(n)
**Amortized
Binary heap wins for most cases.
Python's heapq
Built-in. Min heap by default.
import heapq
pq = []
# Insert
heapq.heappush(pq, (1, "Low"))
heapq.heappush(pq, (5, "High"))
# Peek
if pq:
print(pq[0]) # (1, "Low")
# Extract
priority, item = heapq.heappop(pq)
For max heap, negate priorities:
pq = []
heapq.heappush(pq, (-5, "High"))
heapq.heappush(pq, (-1, "Low"))
priority, item = heapq.heappop(pq)
print(-priority, item) # 5, "High"
Custom Priority Queue with heapq
Complete implementation with tie-breaking:
import heapq
class PriorityQueue:
def __init__(self, max_heap=False):
self.heap = []
self.counter = 0
self.max_heap = max_heap
def push(self, priority, item):
if self.max_heap:
priority = -priority
heapq.heappush(self.heap, (priority, self.counter, item))
self.counter += 1
def pop(self):
if not self.heap:
raise IndexError("Empty queue")
priority, _, item = heapq.heappop(self.heap)
if self.max_heap:
priority = -priority
return priority, item
def peek(self):
if not self.heap:
return None
priority, _, item = self.heap[0]
if self.max_heap:
priority = -priority
return priority, item
def is_empty(self):
return len(self.heap) == 0
def __len__(self):
return len(self.heap)
# Usage
pq = PriorityQueue(max_heap=True)
pq.push(5, "Medium")
pq.push(10, "High")
pq.push(1, "Low")
while not pq.is_empty():
priority, item = pq.pop()
print(f"{item}: {priority}")
# Output:
# High: 10
# Medium: 5
# Low: 1
When to Choose What
Binary heap: Default choice. Balanced, simple, efficient.
Unsorted array: Never in practice. Teaching tool only.
Sorted array: Never. Heap beats it on both operations.
BST: When you need additional operations (range queries, ordered iteration). But adds complexity.
Fibonacci heap: Theoretical interest. Too complex for practical use. Dijkstra optimization in academic papers.
The Pattern
Priority queues abstract priority-based access:
- Interface: insert, extract, peek
- Implementation: usually binary heap
- Trade-offs: simplicity vs special requirements
Understanding implementations means knowing when the default (heap) is good enough and when you need something else.
Master heap-based priority queues, and you'll handle 99% of real-world cases.
