heapq --- 堆隊(duì)列算法?

基本示例?

堆排序 可以通過將所有值推入堆中然后每次彈出一個最小值項(xiàng)來實(shí)現(xiàn)。

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(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() 不同的是這個實(shí)現(xiàn)是不穩(wěn)定的。

堆元素可以為元組。 這適用于將比較值（例如任務(wù)優(yōu)先級）與跟蹤的主記錄進(jìn)行賦值的場合:

>>> 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')

優(yōu)先隊(duì)列實(shí)現(xiàn)說明?

優(yōu)先隊(duì)列 是堆的常用場合，并且它的實(shí)現(xiàn)包含了多個挑戰(zhàn)：

• 排序穩(wěn)定性：你該如何令相同優(yōu)先級的兩個任務(wù)按它們最初被加入時的順序返回？

• 如果優(yōu)先級相同且任務(wù)沒有默認(rèn)比較順序，則 (priority, task) 對的元組比較將會中斷。

• 如果任務(wù)優(yōu)先級發(fā)生改變，你該如何將其移至堆中的新位置？

• 或者如果一個掛起的任務(wù)需要被刪除，你該如何找到它并將其移出隊(duì)列？

針對前兩項(xiàng)挑戰(zhàn)的一種解決方案是將條目保存為包含優(yōu)先級、條目計數(shù)和任務(wù)對象 3 個元素的列表。 條目計數(shù)可用來打破平局，這樣具有相同優(yōu)先級的任務(wù)將按它們的添加順序返回。 并且由于沒有哪兩個條目計數(shù)是相同的，元組比較將永遠(yuǎn)不會直接比較兩個任務(wù)。

不可比較任務(wù)問題的另一種解決方案是創(chuàng)建一個忽略任務(wù)條目并且只比較優(yōu)先級字段的包裝器類:

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

其余的挑戰(zhàn)主要包括找到掛起的任務(wù)并修改其優(yōu)先級或?qū)⑵渫耆瞥?找到一個任務(wù)可使用一個指向隊(duì)列中條目的字典來實(shí)現(xiàn)。

移除條目或改變其優(yōu)先級的操作實(shí)現(xiàn)起來更為困難，因?yàn)樗鼤茐亩呀Y(jié)構(gòu)不變量。 因此，一種可能的解決方案是將條目標(biāo)記為已移除，再添加一個改變了優(yōu)先級的新條目:

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
counter = itertools.count()     # unique sequence count

def add_task(task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

理論?

堆是通過數(shù)組來實(shí)現(xiàn)的，其中的元素從 0 開始計數(shù)，對于所有的 k 都有 a[k] <= a[2*k+1] 且 a[k] <= a[2*k+2]。 為了便于比較，不存在的元素被視為無窮大。 堆最有趣的特性在于 a[0] 總是其中最小的元素。

上面的特殊不變量是用來作為一場錦標(biāo)賽的高效內(nèi)存表示。 下面的數(shù)字是 k 而不是 a[k]:

                               0

              1                                 2

      3               4                5               6

  7       8       9       10      11      12      13      14

15 16   17 18   19 20   21 22   23 24   25 26   27 28   29 30

在上面的樹中，每個 k 單元都位于 2*k+1 和 2*k+2 之上。 體育運(yùn)動中我們經(jīng)常見到二元錦標(biāo)賽模式，每個勝者單元都位于另兩個單元之上，并且我們可以沿著樹形圖向下追溯勝者所遇到的所有對手。 但是，在許多采用這種錦標(biāo)賽模式的計算機(jī)應(yīng)用程序中，我們并不需要追溯勝者的歷史。 為了獲得更高的內(nèi)存利用效率，當(dāng)一個勝者晉級時，我們會用較低層級的另一條目來替代它，因此規(guī)則變?yōu)橐粋€單元和它之下的兩個單元包含三個不同條目，上方單元“勝過”了兩個下方單元。

如果此堆的不變性質(zhì)始終受到保護(hù)，則序號 0 顯然是總的贏家。 刪除它并找出“下一個”贏家的最簡單算法方式是將某個輸家（讓我們假定是上圖中的 30 號單元）移至 0 號位置，然后將這個新的 0 號沿樹下行，不斷進(jìn)行值的交換，直到不變性質(zhì)得到重建。 這顯然會是樹中條目總數(shù)的對數(shù)。 通過迭代所有條目，你將得到一個 O(n log n) 復(fù)雜度的排序。

此排序有一個很好的特性就是你可以在排序進(jìn)行期間高效地插入新條目，前提是插入的條目不比你最近取出的 0 號元素“更好”。 這在模擬上下文時特別有用，在這種情況下樹保存的是所有傳入事件，“勝出”條件是最小調(diào)度時間。 當(dāng)一個事件將其他事件排入執(zhí)行計劃時，它們的調(diào)試時間向未來方向延長，這樣它們可方便地入堆。 因此，堆結(jié)構(gòu)很適宜用來實(shí)現(xiàn)調(diào)度器，我的 MIDI 音序器就是用的這個 :-)。

用于實(shí)現(xiàn)調(diào)度器的各種結(jié)構(gòu)都得到了充分的研究，堆是非常適宜的一種，因?yàn)樗鼈兊乃俣认喈?dāng)快，并且?guī)缀跏呛愣ǖ?，最壞的情況與平均情況沒有太大差別。 雖然還存在其他總體而言更高效的實(shí)現(xiàn)方式，但其最壞的情況卻可能非常糟糕。

堆在大磁盤排序中也非常有用。 你應(yīng)該已經(jīng)了解大規(guī)模排序會有多個“運(yùn)行輪次”（即預(yù)排序的序列，其大小通常與 CPU 內(nèi)存容量相關(guān)），隨后這些輪次會進(jìn)入合并通道，輪次合并的組織往往非常巧妙 1。 非常重要的一點(diǎn)是初始排序應(yīng)產(chǎn)生盡可能長的運(yùn)行輪次。 錦標(biāo)賽模式是達(dá)成此目標(biāo)的好辦法。 如果你使用全部有用內(nèi)存來進(jìn)行錦標(biāo)賽，替換和安排恰好適合當(dāng)前運(yùn)行輪次的條目，你將可以對于隨機(jī)輸入生成兩倍于內(nèi)存大小的運(yùn)行輪次，對于模糊排序的輸入還會有更好的效果。

另外，如果你輸出磁盤上的第 0 個條目并獲得一個可能不適合當(dāng)前錦標(biāo)賽的輸入（因?yàn)槠渲狄皠龠^”上一個輸出值），它無法被放入堆中，因此堆的尺寸將縮小。 被釋放的內(nèi)存可以被巧妙地立即重用以逐步構(gòu)建第二個堆，其增長速度與第一個堆的縮減速度正好相同。 當(dāng)?shù)谝粋€堆完全消失時，你可以切換新堆并啟動新的運(yùn)行輪次。 這樣做既聰明又高效！

總之，堆是值得了解的有用內(nèi)存結(jié)構(gòu)。 我在一些應(yīng)用中用到了它們，并且認(rèn)為保留一個 'heap' 模塊是很有意義的。 :-)

備注

1

當(dāng)前時代的磁盤平衡算法與其說是巧妙，不如說是麻煩，這是由磁盤的尋址能力導(dǎo)致的結(jié)果。 在無法尋址的設(shè)備例如大型磁帶機(jī)上，情況則相當(dāng)不同，開發(fā)者必須非常聰明地（極為提前地）確保每次磁帶轉(zhuǎn)動都盡可能地高效（就是說能夠最好地加入到合并“進(jìn)程”中）。 有些磁帶甚至能夠反向讀取，這也被用來避免倒帶的耗時。 請相信我，真正優(yōu)秀的磁帶機(jī)排序看起來是極其壯觀的，排序從來都是一門偉大的藝術(shù)！ :-)