heapq --- 堆積佇列 (heap queue) 演算法

原始碼：Lib/heapq.py

這個模組實作了堆積佇列 (heap queue) 演算法，亦被稱為優先佇列 (priority queue) 演算法。

最小堆積 (min-heap) 是一顆二元樹，樹上每個父節點的值都小於或等於其子節點的值。我們將這種情況稱為堆積性質不變 (heap invariant)。

對於最小堆積，此實作使用串列，對於所有存在的被比較元素，k 都滿足heap[k]<=heap[2*k+1] 和heap[k]<=heap[2*k+2]。元素是從零開始計數。最小堆積有一個有趣的性質：最小的元素永遠在根節點heap[0]。

Max-heaps satisfy the reverse invariant: every parent node has a valuegreater than any of its children. These are implemented as lists for whichmaxheap[2*k+1]<=maxheap[k] andmaxheap[2*k+2]<=maxheap[k] for allk for which the compared elements exist.The root,maxheap[0], contains thelargest element;heap.sort(reverse=True) maintains the max-heap invariant.

heapq API 與教科書上的堆積演算法在兩個方面不同：(a) 我們使用從零開始的索引。這使得節點索引和其子節點索引之間的關係變得不那麼明顯，但由於 Python 使用從零開始的索引，所以這樣更適合。(b) 教科書通常專注於最大堆積，因為它們適合原地排序。我們的實作偏向於最小堆積，因為它們更符合 Python串列。

這兩個特性使得可以將堆積視為一個普通的 Python 串列而不會有意外：heap[0] 是最小的元素，而heap.sort() 維持堆積性質不變！

Likelist.sort(), this implementation uses only the< operatorfor comparisons, for both min-heaps and max-heaps.

In the API below, and in this documentation, the unqualified termheapgenerally refers to a min-heap.The API for max-heaps is named using a_max suffix.

要建立一個堆積，使用初始化為[] 的串列，或者分別使用heapify() 或heapify_max() 函式將現有的串列轉換為最小堆積或最大堆積。

提供了以下針對最小堆積的函式：

heapq.heappush(heap,item)

將值item 推入heap，並維持最小堆積性質不變。

heapq.heappop(heap)

從heap 取出並回傳最小的元素，維持最小堆積性質不變。如果堆積為空，會引發IndexError。若要在不取出的情況下存取最小元素，請使用heap[0]。

heapq.heappushpop(heap,item)

將item 放入 heap ，接著從heap 取出並回傳最小的元素。這個組合函式比呼叫heappush() 之後呼叫heappop() 更有效率。

heapq.heapify(x)

在線性時間內將串列x 原地轉換為最小堆積。

heapq.heapreplace(heap,item)

從heap 取出並回傳最小的元素，接著將新的item 放進heap。heap 的大小不會改變。如果 heap 是空的會產生IndexError 錯誤。

這個一次完成的操作會比呼叫heappop() 之後呼叫heappush() 更有效率，並在維護 heap 的大小不變時更為適當，取出/放入的組合函式一定會從 heap 回傳一個元素並用item 取代他。

函式的回傳值可能會大於被加入的item 。如果這不是你期望發生的，可以考慮使用heappushpop() 替代，他會回傳 heap 的最小值和item 兩個當中比較小的那個，並將大的留在 heap 內。

提供以下針對最大堆積的函式：

heapq.heapify_max(x)

在線性時間內將串列x 原地轉換為最大堆積。

在 3.14 版被加入.

heapq.heappush_max(heap,item)

將值item 推入最大堆積heap，維持最大堆積性質不變。

在 3.14 版被加入.

heapq.heappop_max(heap)

從最大堆積heap 取出並回傳最大的元素，維持最大堆積性質不變。如果最大堆積為空，會引發IndexError。若要在不取出的情況下存取最大元素，請使用maxheap[0]。

在 3.14 版被加入.

heapq.heappushpop_max(heap,item)

將item 推入 max-heapheap，然後取出並回傳heap 中最大的元素。這個組合動作比先呼叫heappush_max() 再單獨呼叫heappop_max() 更有效率。

在 3.14 版被加入.

heapq.heapreplace_max(heap,item)

從最大堆積heap 取出並回傳最大的元素，同時推入新的item。最大堆積的大小不會改變。如果最大堆積為空，會引發IndexError。

The value returned may be smaller than theitem added. Refer to theanalogous functionheapreplace() for detailed usage notes.

在 3.14 版被加入.

這個模組也提供三個利用 heap 實作的一般用途函式

heapq.merge(*iterables,key=None,reverse=False)

合併多個已排序的輸入並產生單一且已排序的輸出（舉例：合併來自多個 log 檔中有時間戳記的項目）。回傳一個iterator 包含已經排序的值。

和sorted(itertools.chain(*iterables)) 類似但回傳值是一個 iterable ，不會一次把所有資料都放進記憶體中，並且假設每一個輸入都已經（由小到大）排序過了。

有兩個選用參數，指定時必須被當作關鍵字參數指定。

key 參數指定了一個key function 引數，用來從每一個輸入的元素中決定一個比較的依據。預設的值是None （直接比較元素）。

reverse 是一個布林值，如果設定為True ，則輸入的元素將以相反的比較順序進行合併。為了達成類似sorted(itertools.chain(*iterables),reverse=True) 的行為，所有 iterables 必須由大到小排序。

在 3.5 版的變更:加入選用參數key 和reverse 。

heapq.nlargest(n,iterable,key=None)

回傳一個包含資料iterable 中前n 大元素的 list 。如果有指定key 引數，key 會是只有一個引數的函式，用來從每一個在iterable 中的元素提取一個比較的依據（例如key=str.lower ）。效果相當於sorted(iterable,key=key,reverse=True)[:n] 。

heapq.nsmallest(n,iterable,key=None)

回傳一個包含資料iterable 中前n 小元素的 list 。如果有指定key 引數，key 會是只有一個引數的函式，用來從每一個在iterable 中的元素提取一個比較的依據（例如key=str.lower ）。效果相當於sorted(iterable,key=key)[:n] 。

後兩個函式在n 值比較小時有最好的表現。對於較大的n 值，只用sorted() 函式會更有效率。同樣地，當n==1 時，使用內建函式min() 和max() 會有更好的效率。如果需要重複使用這些函式，可以考慮將 iterable 轉成真正的 heap 。

基礎範例

堆積排序 (heapsort) 可以透過將所有的值推入一個 heap，並且從 heap 中一個接一個彈出最小元素來實作：

>>>defheapsort(iterable):...h=[]...forvalueiniterable:...heappush(h,value)...return[heappop(h)foriinrange(len(h))]...>>>heapsort([1,3,5,7,9,2,4,6,8,0])[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

雖然類似sorted(iterable) ，但跟sorted() 不同的是，這個實作不是 stable 的排序。

Heap 中的元素可以是 tuple 。這有利於將要比較的值（例如一個 task 的優先度）和主要資料放在一起排序：

>>>h=[]>>>heappush(h,(5,'write code'))>>>heappush(h,(7,'release product'))>>>heappush(h,(1,'write spec'))>>>heappush(h,(3,'create tests'))>>>heappop(h)(1, 'write spec')

優先佇列實作細節

優先佇列 (priority queue) 是 heap 的常見用途之一，實作優先佇列伴隨著下列挑戰：

  • 排序的穩定性：如何將兩個擁有相同優先次序 (priority) 的 task 按照他們被加入的順序回傳?

  • Tuple的排序在某些情況下會壞掉，例如當 Tuple (priority, task) 的 priorities 相等且 tasks 沒有一個預設的排序時。

  • 當一個 heap 中 task 的 priority 改變時，如何將它移到 heap 正確的位置上?

  • 或者一個還沒被解決的 task 需要被刪除時，要如何從佇列中找到並刪除指定的 task?

一個針對前兩個問題的解法是：儲存一個包含 priority、entry count 和 task 三個元素的 tuple 。兩個 task 有相同 priority 時，entry count 會讓兩個 task 能根據加入的順序排序。因為沒有任何兩個 task 擁有相同的 entry count，所以永遠不會直接使用 task 做比較。

task 無法比較的另一個解決方案是建立一個包裝器類別，該類別忽略 task 項目，只比較優先等級：

fromdataclassesimportdataclass,fieldfromtypingimportAny@dataclass(order=True)classPrioritizedItem:priority:intitem:Any=field(compare=False)

剩下的問題可以藉由找到要刪除的 task 並更改它的 priority 或者直接將它移除。尋找一個 task 可以使用一個 dictionary 指向佇列當中的 entry 。

移除 entry 或更改它的 priority 更為困難，因為這會破壞 heap 的性質。所以一個可行的方案是將原本的 entry 做一個標記表示它已經被刪除，並新增一個擁有新的 priority 的 entry：

pq=[]# 在 heap 中的 entry 串列entry_finder={}# task 對應到 entry 的對映REMOVED='<removed-task>'# 被刪除的 task 的佔位器counter=itertools.count()# 唯一的序列計數defadd_task(task,priority=0):'新增一個 task 或更新一個已存在 task 的 priority'iftaskinentry_finder:remove_task(task)count=next(counter)entry=[priority,count,task]entry_finder[task]=entryheappush(pq,entry)defremove_task(task):'將一個已存在的 task 標記為 REMOVED。如果找不到會引發 KeyError。'entry=entry_finder.pop(task)entry[-1]=REMOVEDdefpop_task():'移除並回傳最低 priority 的 task。如果 heap 是空的會引發 KeyError。'whilepq:priority,count,task=heappop(pq)iftaskisnotREMOVED:delentry_finder[task]returntaskraiseKeyError('從空的優先佇列中 pop 出元素')

原理

Heap 是一個陣列對於所有從0開始的 indexk 都存在性質a[k]<=a[2*k+1] 和a[k]<=a[2*k+2] 。為了方便比較，不存在的元素被視為無限大。Heap 的一個有趣的性質是：a[0] 永遠是最小的元素。

上述乍看之下有些奇怪的不變式，是為了實作一個對記憶體來說有效率的方法，其表示方式如同錦標賽一般。下列的數字為k，而不是a[k]：

0123456789101112131415161718192021222324252627282930

在上面的樹當中，每個單元k 都會位在2*k+1 與2*k+2 上方。如同體育賽事常見的錦標賽般，每個單元可視為其下方兩個單元當中的贏家，我們可以透過追溯整棵樹來找到該贏家曾經對戰過的所有對手。然而，在許多電腦應用中，我們不需要追溯贏家的完整對戰歷史。為了能更有效率地使用記憶體，當一個贏家晉級勝出時，我們用下方較低層級的另一個項目來取代它，至此規則變為一個單元以及它下方兩個單元，包含三個不同項目，但是最上方的單元「勝過」下方兩個單元。

如果能確保滿足這個 heap 的不變式，那麼索引 0 顯然是最終的贏家。移除並找到「下一個」贏家最簡單的演算法為：將一個輸家（例如上圖中的單元 30）移動到位置 0，然後從新的位置 0 不斷與下方的位置交換值來向下傳遞，直到滿足不變式為止。這個過程的複雜度顯然是樹的節點數目的對數級別。透過對所有項目疊代，可以得到一個複雜度為O(n logn) 的排序。

這種排序有個好處，只要插入的項目沒有「贏過」你最後提取、索引為 0 的元素，你就可以在排序進行的同時有效率地插入新項目。這在模擬情境當中特別有用，其中樹能夠保存所有輸入事件，而「贏」意味著最小排程時間。當一個事件排程其它事件的執行時，因這些事件仍在等待進行，所以很容易將它們插入 heap 當中。因此， heap 是一個實現排程器的優秀資料結構（這就是我用以實作 MIDI 編曲器的方法 :-）。

多種用於實作排程器的結構現今已被廣泛研究，heap 對此非常有用，因為它們速度相當快，且速度幾乎不受其他因素影響，最壞情況與平均狀況差異無幾。也有其它整體說來更有效率的方法，然而它們的最壞情況可能會非常糟糕。

Heap 在為儲存於硬碟上的大量資料進行排序也非常有用。你可能已經知道，大量資料排序涉及 "runs" 的產生（也就是預先排序的序列，其大小通常與 CPU 記憶體的大小有關），之後再對這些 run 合併，而這些合併的過程通常相當巧妙[1]。很重要的一點是，初始排序產生的 run 越長越好。錦標賽是達成這一點的好方法，若你用所有可用記憶體來舉行一場錦標賽，並透過替換與向下交換來處理所有適配目前 run 的值，那麼對於隨機產生的輸入，將可以產生長度兩倍於記憶體大小的 run。對於已模糊排序過的輸入，效果更好。

此外，若你將索引為 0 的項目輸出至磁碟，並取得一個無法適配目前錦標賽的輸入（因為該值「勝過」最後的輸出值），則該輸入值就無法插入至 heap 當中，因此 heap 的大小會減小。釋放出來的記憶體可以巧妙地立即再被運用，逐步建構出第二個 heap，其大小增加的速度會與第一個 heap 減少的速度一致。當第一個 heap 完全消失時，你可以切換至第二個 heap 開啟一個新 run 。這真是個聰明且相當有效率的做法！

總結來說，heap 是值得了解的有用記憶體結構。我在一些應用中使用它們，我認為能有一個 'heap' 模組是很棒的。:-)

註解

[1]

現今的磁碟平衡演算法因為硬碟查找能力而更加複雜難解。在沒有查找功能的裝置如大型磁帶機，狀況又不一樣了，人們必須機智地確保（遠遠提前）每次於磁帶上移動都盡可能是最有效率的（也就是盡可能更好地「推進」合併的過程）。有些磁帶甚至能夠向後讀取，這也被用來避免倒轉的時間。相信我，真正優秀的磁帶排序看起來相當壯觀！排序一直以來都是一門偉大的藝術！:-)