堆排序 ( Heap Sort ) - _jaskell - 博客大巴

今天来了解一下堆排序的问题，“堆”是个很有趣的结构。通常“堆”是通过数组来实现的，这样可以利用数组的特点快速定位指定索引的元素。而对“堆”的操作中定位某个元素简直就是家常便饭。为什么这么说呢，我们来看看“堆”的定义：

在起始索引为 0 的“堆”中：
1) 堆的根节点将存放在位置 0
2) 节点 i 的左子节点在位置 2 * i + 1
3) 节点 i 的右子节点在位置 2 * i + 2
4) 节点 i 的父节点在位置 floor( (i - 1) / 2 )   : 注 floor 表示“取整”操作

在起始索引为 1 的“堆”中：
1) 堆的根节点将存放在位置 1
2) 节点 i 的左子节点在位置 2 * i
3) 节点 i 的右子节点在位置 2 * i + 1
4) 节点 i 的父节点在位置 floor( i / 2 )           : 注 floor 表示“取整”操作

以下是一个“堆”的图例：

因此，当我们得知某个节点的索引为 i 时，可以通过以下“堆”的定义很容易地定位到它的子节点和父节点。

不仅如此，“堆”还有个特性：每个节点的键值一定总是大于（或小于）它的父节点。如果我们修改了某个节点的键值，这样就会破坏“堆”的这个特性，因此这时我们就要根据该节点的新键值与它的父节点和子节点的键值进行比较，对该节点进行“上移”或者“下移”操作，使之能够重新放置到合适位置。

这里我们了解一下“堆”的这两个很重要的操作：“上移”和“下移”。

这里我们假设这是一个“最大堆”，即“堆”的根节点保存的是键值最大的节点。即“堆”中每个节点的键值都总是大于它的子节点。

当某节点的键值大于它的父节点时，这时我们就要进行“上移”操作，即我们把该节点移动到它的父节点的位置，而让它的父节点到它的位置上，然后我们继续判断该节点，直到该节点不再大于它的父节点为止才停止“上移”。

现在我们再来了解一下“下移”操作。当我们把某节点的键值改小了之后，我们就要对其进行“下移”操作。首先，我们要取该节点的两个左右子节点中键值较大的一个，与该节点进行比较，如果该节点小于它的这个子节点，就把该节点移动到它的子节点的位置，而让它原来的子节点升到它的位置上。然后我们继续判断该节点，直到该节点不再小于它的左右子节点为止才停止“下移”。

现在我们再来看看如何把一个无序的序列建立成为一个“堆”？

根据“堆”的特性，我们可以从无序序列的最后一个节点的父节点开始，对其进行“下移”操作，直到序列的第一个节点为止。这样就可以保证每个节点都在合适的位置上，也就建立起了一个“堆”。

但是“堆”并不是一个完全排序的序列，因为“堆”只保证了父节点与子节点的位置关系，但并不保证左右子节点的位置关系。那么我们如何进行“堆排序”呢？

由于一个“堆”的根节点必然是整个序列中最大的元素，因此对于一个排序的序列而言，每个“堆”中我们只能得到一个有效的元素。如果我们拿掉根节点，再对剩下的序列重新排列组成一个“堆”，反复如此，我们就可以依次得到一个完整的排序序列了。

当然，为了简化操作，每次我们只需要把根节点与最后一个位置的节点交换，然后把最后一个位置排除之外，对根节点进行“下移”操作即可。

有了这些基础，我们实现一个“堆排序”就不再困难了。以下是我用 Python 实现的堆排序：

# -*- coding: gb2312 -*-

import os, sys
import random
import heapq

# 以起始索引 0 开始的"堆"的每个节点 i ，其左子节点为 2*i+1， 其右子节点为 2*i+2

# 上移操作
def siftUp(slist, idx):
    while idx > 0:                                                  # 范围 [1, idx]
        pidx = (idx - 1) // 2
        if slist[idx] > slist[pidx]:
            slist[idx], slist[pidx] = slist[pidx], slist[idx]
            idx = pidx
        else:
            break
    return slist

# 下移操作
def siftDown(slist, idx, num):
    last = num // 2 - 1                                             # 计算最后一个节点的父节点索引
    while idx <= last:                                              # 范围 [idx, last]
        lidx = 2 * idx + 1
        ridx = 2 * idx + 2
        if ridx < num and slist[lidx] < slist[ridx]:
            lidx = ridx
        if slist[lidx] > slist[idx]:
            slist[idx], slist[lidx] = slist[lidx], slist[idx]
            idx = lidx
        else:
            break
    return slist

# 插入操作
def insertItem(slist, val):
    slist.append(val)
    siftUp(slist, len(val)-1)

# 删除操作
def deleteItem(slist, idx):
    compare = slist[idx] < slist[-1]
    del slist[-1]
    if compare:
        siftUp(slist, idx)
    else:
        siftDown(slist, idx)

# 建立堆
def makeHeap(slist):
    num  = len(slist)
    last = num // 2 - 1
    for idx in xrange(last, -1, -1):
        siftDown(slist, idx, num)

# 堆排序
def heapSort(slist):
    makeHeap(slist)
    num = len(slist) - 1
    for n in xrange(num, 0, -1):
        slist[n], slist[0] = slist[0], slist[n]
        siftDown(slist, 0, n)

在各种语言中当然都不需要我们再去自己实现排序了。在 Python 中 list 就自带有 sort 函数。而且 heapq 模块中也有 heapify 函数再建立一个“堆”， heappush 函数向“堆”中添加一个元素， heappop 函数取出“堆”的根节点。因此我们也可以利用 Python 语言自带的这些功能来实现“堆排序”：

def inHeapSort3(slist):
    heapq.heapify(slist)
    heap = []
    while slist:
        heap.append(heapq.heappop(slist))
    slist[:] = heap

有了这些，我们再来做个小小的测试：

# 利用 Python 提供的函数 heappush 建立堆，然后进行堆排序
def inHeapSort1(slist):
    heap = []
    for item in slist:
        heapq.heappush(heap, item)

    xlist = []
    while heap:
        xlist.append(heapq.heappop(heap))
    slist[:] = xlist

# 利用 Python 提供的函数 heappush 建立堆，然后进行堆排序
def inHeapSort2(slist):
    heap = []
    for item in slist:
        heapq.heappush(heap, item)

    for i in xrange(len(slist)):
        slist[i] = heapq.heappop(heap)

# 利用 Python 提供的函数 heapify 建立堆，然后进行堆排序
def inHeapSort3(slist):
    heapq.heapify(slist)
    heap = []
    while slist:
        heap.append(heapq.heappop(slist))
    slist[:] = heap

# 利用自身的函数 sort 进行排序
def selfSort(slist):
    slist.sort()

# 测试
if __name__ == '__main__':
    n       = 1000                    # 测试样例所用无序列表大小
    loopnum = 10                      # 每个样例循环测试次数
   
    xlist = range(1, n+1)
    slist = []
    slist[:] = xlist
    random.shuffle(slist)             # 建立无序列表

    exams = ['selfSort', 'heapSort', 'inHeapSort1', 'inHeapSort2', 'inHeapSort3']
    from timeit import Timer
    for i in xrange(len(exams)):
        tlist    = []
        tlist[:] = slist              # 复制一份作测试使用
        stmt = exams[i] + "(tlist); assert(xlist == tlist)";
        setup = "from __main__ import heapSort, xlist, slist, tlist, " + exams[i]
        t = Timer(stmt, setup)
        print exams[i].ljust(20) + " = " + str(t.timeit(loopnum))

在我的机器上测试结果如下：

                       n = 10               n = 100             n = 1000            n = 10000           n = 100000
selfSort             = 0.000313168293736    0.000351441314469   0.000947885834652   0.0704061549722     0.119555977086
heapSort             = 0.000136050810927    0.0024754542826     0.0552084133598     0.572977571172      7.44314486897
inHeapSort1          = 3.6596830044e-005    0.000358984172569   0.00608764521748    0.0328801565562     0.358665416973
inHeapSort2          = 5.89460392312e-005   0.000482463553329   0.00692127072016    0.0551254419207     0.585413509259
inHeapSort3          = 2.68190510246e-005   0.000174603196775   0.00331438772246    0.0243371713444     0.270264796224

可以看到一个很有趣的现象，当无序序列元素很少时，list 自带的 sort 函数与我实现的“堆排序”相比并不占优势，只有当元素过万时才显现出强大的优势。当然相对于同样是内部函数 heapify + heappop 实现的“堆排序”而言，它们则相差无几。

如果只了解“堆排序”的实现，显然没有什么乐趣。那么它还有什么实际的应用没有？这里举个小例子：

假如某一个文件中包含了1亿个随机的整数，要求你快速的找出最小的100万个数？

这类问题该如何解决呢？其实使用“堆”就很容易解决。以下是我的一个思路：
1) 首先我们用可以保存100万个数的数组读入这些整数并建立成一个“堆”。
2) 然后读入一个整数，如果该整数大于等于“堆”的根节点，则丢弃该整数，进入步骤(3)。如果该整数小于“堆”的根节点，则用此整数替换根节点，并对根节点进行“下移”操作以重建“堆”。
3) 重复步骤(2)直到所有整数都被处理。最后这个“堆”中保存的就是这最小的100万个数。