2.38. 優先佇列

優先佇列是一種容器，它無須對所有項目進行排序，就能隨時知道哪個項目最小（或最大）。heapq 在 MicroPython 中以一個普通的 list 作為儲存空間，並使用一種稱為 min-heap（最小堆積） 的樹狀演算法來實作優先佇列，使最小的項目始終位於索引 0。

2.38.1. 概念模型

min-heap 是一種經過排列的 list，使得每個索引處的項目都小於或等於索引 2*i + 1 與 2*i + 2 處的項目。這種排列方式讓三項操作變得低成本：

  • 找出最小的項目 -- 它位於 heap[0]。只需一次索引查找，不必掃描。

  • 新增項目 -- 它會在樹中向上浮動，每一層與一個父節點比較一次。一千個項目約需十次比較；一百萬個約需二十次。

  • 移除最小的項目 -- 以相同的方式反向進行，比較次數同樣不多。

這個 list 並非 排序過的。逐一走訪它會以任意順序得到項目。只有 heap[0] 保證是最小值。

2.38.2. MicroPython 上的三個函式

MicroPython 的 heapq 正好提供三個函式：

這就是它的全部功能。CPython 的 heappushpop、heapreplace、nlargest、nsmallest 與 merge 並不 存在；下面的範例會示範如何手動建立這些常見的功能。

從頭開始建立:

>>> import heapq
>>> heap = []
>>> heapq.heappush(heap, 5)
>>> heapq.heappush(heap, 1)
>>> heapq.heappush(heap, 3)
>>> heap
[1, 5, 3]
>>> heapq.heappop(heap)
1
>>> heapq.heappop(heap)
3
>>> heapq.heappop(heap)
5

從既有資料開始:

>>> samples = [5, 1, 3, 2, 4]
>>> heapq.heapify(samples)
>>> samples
[1, 2, 3, 5, 4]
>>> heapq.heappop(samples)
1

heapify 對每個項目只處理一次，明顯比逐一推入相同的項目更快。

2.38.3. 串流中的前 N 大

優先佇列的一個自然用途：在只能讀取一次的串流中，追蹤其中 N 個最大的項目。

訣竅是維持一個大小為 N 的 min-heap。每個新值都會被推入；若堆積成長超過 N，就丟棄最小的。最小值永遠位於 heap[0]，因此只需一次比較即可決定保留或捨棄每個進入的項目:

def top_n_brightest(readings, n):
    heap = []
    for r in readings:
        if len(heap) < n:
            heapq.heappush(heap, r)
        elif r > heap[0]:
            heapq.heappop(heap)
            heapq.heappush(heap, r)
    return sorted(heap, reverse=True)

堆積永遠不會成長超過 N，因此無論串流多長，每個進入的值所花的比較次數都相同。最後的 sorted 是唯一一個成本直接取決於 N 的步驟。

2.38.4. 排程事件

由 (deadline, payload) 元組組成的優先佇列是事件排程器的標準形式。元組會以字典順序比較，因此 deadline 最小的元組會自動浮到最上方:

import heapq
import time

queue = []
heapq.heappush(queue, (time.time() + 5, 'check sensors'))
heapq.heappush(queue, (time.time() + 1, 'blink LED'))
heapq.heappush(queue, (time.time() + 3, 'send heartbeat'))

while queue:
    deadline, task = queue[0]
    now = time.time()
    if now < deadline:
        time.sleep(deadline - now)
    heapq.heappop(queue)
    run(task)

若兩個事件共用同一個 deadline，元組比較會落到第二個元素。對於本身不易比較的 payload，可以在元組的第二個元素加上一個計數器，以強制穩定的排序:

counter = 0
def schedule(deadline, task):
    global counter
    counter += 1
    heapq.heappush(queue, (deadline, counter, task))

現在 deadline 最小者優先的行為維持不變，平手時以計數器決勝，因此較早排程的任務會先執行，而且 task 不必是可比較的。

2.38.5. heapq 不提供的功能

有幾種看似應該可行、實際上卻不行的模式：

  • 以 for x in heap 走訪 不會 得到排序過的輸出 -- list 底層的順序是堆積的樹狀佈局。若要依序消耗堆積，請持續 heappop 直到清空。

  • 無法 變更 已經在堆積中的項目的優先順序。標準的變通做法是將新的優先順序作為一筆新項目推入，並把舊的標記為已取消（當它彈出時略過它）。

  • 堆積儲存的是其項目的參考，而非副本。在項目被推入後對其進行變動，可能會破壞堆積的不變性。請使用不可變的值（數字、元組）作為比較的鍵。

在普通 list 上實作的 min-heap，是小型嵌入式裝置上回答任何「哪個最小」或「哪個最緊急」問題的合適工具。三個函式、一個 list，其餘的就交給堆積處理。