3.3 優先權佇列的觀念#
概念說明#
普通佇列「先來先服務」,但很多場景需要「最急的先做」:急診室先診治重症病患、作業系統先執行高優先權行程、Dijkstra 演算法先展開距離最近的節點。優先權佇列(priority queue) 的 ADT 是:
push(x, priority):加入元素。pop():取出並回傳優先權最高的元素(不論先後)。peek():看最高優先權元素。
支援優先權佇列的結構有好幾種:有序陣列(pop O(1) 但 push O(n))、無序陣列(push O(1) 但 pop O(n))、二元搜尋樹(皆 O(log n))、以及最常用的二元堆積(binary heap)——push、pop 都是 O(log n)、peek O(1),且可用陣列緊湊儲存。本章先建立觀念與用法,第7章再深入堆積的內部運作與 heap sort。
Python 的 heapq#
heapq 是最小堆積:heappush / heappop 都取出最小元素。要「最大優先」可存負數或自訂比較。
import heapq
tasks = [] # (優先權, 順序, 任務) 三元組
seq = 0 # 順序號碼:同優先權時保持 FIFO
def add_task(task, prio):
global seq
heapq.heappush(tasks, (prio, seq, task))
seq += 1
add_task("寫報告", 3)
add_task("滅火", 1)
add_task("回訊息", 5)
add_task("二級火警", 1)
while tasks:
prio, _, task = heapq.heappop(tasks)
print(f"執行:{task}(優先權 {prio})")
# 滅火 → 二級火警(同為1,先來先做)→ 寫報告 → 回訊息
用「病患排程」自製優先權佇列包裝#
import heapq
class PriorityQueue:
def __init__(self):
self._heap = []
self._seq = 0 # 解決同優先權比較問題,並保持 FIFO
def push(self, item, priority):
heapq.heappush(self._heap, (priority, self._seq, item))
self._seq += 1
def pop(self):
if not self._heap:
raise IndexError("佇列為空")
return heapq.heappop(self._heap)[2]
def peek(self):
return self._heap[0][2]
def __len__(self):
return len(self._heap)
pq = PriorityQueue()
pq.push("感冒", 5)
pq.push("車禍大出血", 1)
pq.push("骨折", 2)
pq.push("發燒", 4)
print([pq.pop() for _ in range(len(pq))])
# ['車禍大出血', '骨折', '發燒', '感冒']
三元組 (priority, seq, item) 是實務上的關鍵技巧:seq 保證當優先權相同時按插入順序排列,也避免 Python 無法比較兩個自訂物件而當機。
三種實現的取捨#
| 實現 | push | pop | peek | 適合 |
|---|---|---|---|---|
| 無序陣列 | O(1) | O(n) | O(n) | push 遠多於 pop |
| 有序陣列 | O(n) | O(1) | O(1) | pop 遠多於 push |
| 二元堆積 | O(log n) | O(log n) | O(1) | 一般情況、兩者皆多 |
堆積是「中間路線」:兩種操作都不超過 O(log n),且不像有序陣列要在 push 時排序整條資料。
複雜度分析#
heapq.heappush / heappop 都是 O(log n),heap[0] 是 O(1)。需要「取出前 k 大/小」的問題(Top-K)用大小為 k 的堆積可做到 O(n log k),遠勝全排序的 O(n log n)。
練習#
- 用
heapq實作「合併 k 條已排序串列」:每條串列各取當前最小放入堆積,反覆 pop 最小者並補上該串列的下一個元素。分析複雜度(應為 O(n log k))。 - 無序陣列實現的優先權佇列在什麼資料分布下反而比堆積快?請用具體數字估計。
- 修改
PriorityQueue,支援remove(item):從堆積中移除指定任務(提示:惰性刪除——標記無效,pop 時跳過)。