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

基本示例?

堆排序 可以通過將所有值推入堆中然后每次彈出一個(gè)最小值項(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() 不同的是這個(gè)實(shí)現(xiàn)是不穩(wěn)定的。

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

>>> 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ì)列 是堆的常用場(chǎng)合，并且它的實(shí)現(xiàn)包含了多個(gè)挑戰(zhàn)：

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

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

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

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

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

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

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)先級(jí)或?qū)⑵渫耆瞥?找到一個(gè)任務(wù)可使用一個(gè)指向隊(duì)列中條目的字典來實(shí)現(xiàn)。

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

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 開始計(jì)數(shù)，對(duì)于所有的 k 都有 a[k] <= a[2*k+1] 且 a[k] <= a[2*k+2]。 為了便于比較，不存在的元素被視為無(wú)窮大。 堆最有趣的特性在于 a[0] 總是其中最小的元素。

上面的特殊不變量是用來作為一場(chǎng)錦標(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

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

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

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

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

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

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

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

備注

1

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