优先队列与堆数据结构详解

一、什么是优先队列？

1. 优先队列的概念

优先队列其实就是队列的一种，不过优先队列是区别于普通队列的。普通队列是一种先进先出，后进后出的数据结构，优先队列和普通队列的区别就在于，出队的顺序和入队的顺序无关，是和优先级息息相关的。

在这个场景中，由于现在计算机都是多任务执行的，我们的操作系统会动态的选择优先级最高的任务执行；因为我们无法准确预估有多少的任务需要处理，所以我们的操作系统只能根据优先级来进行资源的分配。

2. 优先队列的底层实现方式

• 普通线性结构（数组 链表）：入队其实就是一个添加的操作，所以时间复杂度为O(1)；但是出队的操作由于需要执行查询的操作所以时间复杂度是O(n)的。

• 顺序线性结构：与普通线性结构相比，出队的操作是O(1)的，但是入队是O(n)的，因为在入队的时候我们需要对新插入的元素进行比较以便于确定新插入元素的位置。

• 堆：堆是一种高效的数据结构，用堆来实现优先队列的话它的入队和出队时间复杂度都是O(log n)的级别，这里说的O(log n)是指在最坏的情况下都是这个级别的。

二、堆的基本表示

一个堆其实也就是一棵树，这里我们所要使用的就是二叉堆。

说白了，二叉堆就是满足一些特殊性质的二叉树。

1. 二叉堆的特殊性质

性质一：二叉堆是一颗完全二叉树

上图是一个满二叉树，在介绍二分搜索树的笔记里有关于满二叉树的介绍：满二叉树表示除了叶子节点，其他的各个节点都有左右两个节点；但是满二叉树并不是完全二叉树。

完全二叉树：把元素数据排列成树的形状就是完全二叉树；所以对于完全二叉树来说，树的右下角可能会有缺失甚至是空的。

性质二：在堆中某个节点的值总是不大于其父节点的值

注意：从图中可以看到二叉堆中的某些节点甚至小于叶子节点的值，但是这并不影响二叉堆的特性，在二叉堆中节点的值和节点所处的层次之间是没有关系的。

只有同时满足了以上两个特性才能称为二叉堆。

2. 二叉堆的实现方式

方式一：使用数组实现

我们完全可以使用实现二分搜索树的方式来实现二叉堆，但是由于二叉堆是通过把元素数据按照顺序进行排列形成的树的形状，所以我们可以使用数组的方式来实现二叉堆。

通过这个图我们可以清楚的看到二叉堆中的各个元素的层级顺序是怎样的，所有我们可以直接使用数组来实现这样的一个结构。

方式二：节点间的关系表示

当我们使用数组来实现这样一个结构的时候我们怎么去表示一个节点的左右子节点呢？如果我们使用二分搜索树那样的实现方式来实现的话，我们可以通过定义节点的左右子节点来实现，但是在数组中我们显然不能使用节点定义这样的方式。

但是通过观察我们发现了这样的一个规律，就是对于任意个节点来说，它的左节点的索引是其父节点索引的2倍，右节点是其父节点索引的2倍+1：

left_child(i) = 2 * i
right_child(i) = 2 * i + 1

使用这样的方式还有一个好处就是，我们可以很容易的通过任意一个节点找到其父节点，可以直接使用当前节点除以2就可以找到父节点，如果是右节点的话是需要取整的也就是说需要将小数部分抹除掉：

parent(i) = i // 2

这就是使用数组来实现二叉堆的好处，数组的索引之间是存在这样的逻辑关系的。

方式三：索引为0的位置处理

当我们使用数组来实现这样的结构时还有一个问题，就是数组中索引为0的位置该怎么处理？之前在实现循环队列以及链表的时候我们会有意的浪费一个节点，但是在这里我们不用采用这样的方式，我们只需要对我们的节点进行一次偏移的操作就可以了：

parent(i) = (i - 1) // 2
left_child(i) = 2 * i + 1
right_child(i) = 2 * i + 2

这样我们就可以直接使用这个索引为0的位置了。

三、代码实现最大二叉堆

from typing import List, TypeVar, Generic, Optional

T = TypeVar('T')

class Array(Generic[T]):
    def __init__(self, capacity: int = 10):
        self._data: List[Optional[T]] = [None] * capacity
        self._size = 0
        self._capacity = capacity

    def get_size(self) -> int:
        """返回数组中元素个数"""
        return self._size

    def is_empty(self) -> bool:
        """返回数组是否为空"""
        return self._size == 0

    def add_last(self, e: T) -> None:
        """在数组末尾添加元素"""
        if self._size >= self._capacity:
            self._resize(self._capacity * 2)
        self._data[self._size] = e
        self._size += 1

    def get(self, index: int) -> T:
        """获取指定索引的元素"""
        if index < 0 or index >= self._size:
            raise IndexError("Index is out of range")
        return self._data[index]

    def set(self, index: int, e: T) -> None:
        """设置指定索引的元素"""
        if index < 0 or index >= self._size:
            raise IndexError("Index is out of range")
        self._data[index] = e

    def remove_last(self) -> T:
        """删除并返回最后一个元素"""
        if self._size == 0:
            raise RuntimeError("Array is empty")
        ret = self._data[self._size - 1]
        self._data[self._size - 1] = None
        self._size -= 1

        if self._size == self._capacity // 4 and self._capacity // 2 != 0:
            self._resize(self._capacity // 2)
        return ret

    def swap(self, i: int, j: int) -> None:
        """交换两个位置的元素"""
        if i < 0 or i >= self._size or j < 0 or j >= self._size:
            raise IndexError("Index is out of range")
        self._data[i], self._data[j] = self._data[j], self._data[i]

    def _resize(self, new_capacity: int) -> None:
        """调整数组大小"""
        new_data = [None] * new_capacity
        for i in range(self._size):
            new_data[i] = self._data[i]
        self._data = new_data
        self._capacity = new_capacity

class MaxHeap(Generic[T]):
    """最大堆实现"""

    def __init__(self, capacity: int = 10):
        self._data = Array[T](capacity)

    def get_size(self) -> int:
        """返回堆中的元素个数"""
        return self._data.get_size()

    def is_empty(self) -> bool:
        """返回一个boolean值，表示堆是否为空"""
        return self._data.is_empty()

    def _parent(self, index: int) -> int:
        """返回完全二叉树的数组表示中，一个索引的父节点索引值"""
        if index == 0:
            raise ValueError("index-0 doesn't have parent.")
        return (index - 1) // 2

    def _left_child(self, index: int) -> int:
        """返回完全二叉树的数组表示中，一个父节点的左节点的索引值"""
        return index * 2 + 1

    def _right_child(self, index: int) -> int:
        """返回完全二叉树的数组表示中，一个父节点的右节点的索引值"""
        return index * 2 + 2

四、向堆中添加元素和Sift Up

1. 添加过程图示

从树型结构的角度去看，添加一个元素等于是在树中增加一个节点，但是我们的二叉堆底层是使用数组实现的，所以向堆中添加元素实际上操作的是数组。

从这张图片可以看出，添加一个元素就相当于在索引为10的位置插入一个元素，但是这样的操作虽然说满足了完全二叉树的结构，但是并没有满足某个节点的值总是不大于其父节点的值这个特性，所以我们还需要进行调整。

调整也是比较简单的，因为就算是出现需要调整的情况也是发生在新加入的元素与其父节点以及祖先节点的这条链路上，所以我们只要对这一条链路进行调整即可。

所以在这里我们将新加入的数据以及对应的父节点进行比较，然后进行位置的对调即可。

调整完毕以后我们发现还是不能完全满足二叉堆的特性，所以我们继续重复上一步操作进行元素位置的对调。

到这里堆中元素的调整过程就完成了。

这个过程就是Sift Up也就是堆中元素的上浮过程。

2. 代码实现

def add(self, e: T) -> None:
    """向堆中添加一个元素"""
    self._data.add_last(e)
    # 执行sift_up操作
    self._sift_up(self._data.get_size() - 1)

def _sift_up(self, k: int) -> None:
    """元素上浮操作"""
    while k > 0 and self._data.get(self._parent(k)) < self._data.get(k):
        self._data.swap(k, self._parent(k))
        k = self._parent(k)

五、从堆中取出元素和Sift Down

1. 从堆中取出元素图示

由于我们的二叉堆是一个最大二叉堆，所以我们的取出操作只能取出最大的那个值，也就是索引为0的值。

但是取出的最大值同时也是堆中的根节点一旦取出就会导致二叉堆结构的不成立，所以我们需要想办法解决这个问题，第一种方式是将剩下的两个堆融合在一起，但是这样做的话会非常麻烦；所以我们采用其他方法来解决这个问题。

如图所示，我们采用的方式是将最后一个值放到被取出值的位置；这样的话我们的二叉堆的根节点就变成了16，然后我们删除最后一个节点。

这样我们的数组中就没有索引为10的元素了，从元素个数的角度看本次操作是成功的，但是这样操作以后我们的二叉堆的特性就得不到满足了。

这个时候我们就将根节点与其对应的左右子节点进行对比，如果根节点小于左右根节点我们就将根节点与两个子节点中最大的那个进行位置的交换。

在本次示例我们将16与52进行位置上的交换；交换完成以后52这个元素作为根节点就比它的两个子节点都大了。

由于交换完成以后有可能会出现无法满足二叉堆特性的情况所以仍要继续进行交换。

交换完成以后的节点很明显比其子节点要大。

然后我们在将16与其子节点进行对比，由于这一次子节点只有一个我们只能对比一个，对比以后发现子节点是小于它的父节点，同时也是满足二叉堆的特性的就不进行位置上的交换了。

这个过程就叫做Sift Down（数据下沉），通过这个操作我们就完成从堆中取出一个数据的操作。

2. 代码实现

def find_max(self) -> T:
    """查看堆中的最大元素"""
    if self._data.get_size() == 0:
        raise RuntimeError("Can not find_max when heap is empty.")
    return self._data.get(0)

def extract_max(self) -> T:
    """从堆中取出最大的元素"""
    ret = self.find_max()
    self._data.swap(0, self._data.get_size() - 1)
    self._data.remove_last()
    self._sift_down(0)
    return ret

def _sift_down(self, k: int) -> None:
    """元素下沉"""
    while self._left_child(k) < self._data.get_size():
        j = self._left_child(k)

        if (j + 1 < self._data.get_size() and 
            self._data.get(j + 1) > self._data.get(j)):
            j = self._right_child(k)

        if self._data.get(k) >= self._data.get(j):
            break

        self._data.swap(k, j)
        k = j

3. 测试

import random

def test_max_heap():
    n = 1000000

    max_heap = MaxHeap[int]()
    for _ in range(n):
        max_heap.add(random.randint(0, 2**31 - 1))

    arr = []
    for _ in range(n):
        arr.append(max_heap.extract_max())

    for i in range(1, n):
        if arr[i-1] < arr[i]:
            raise ValueError("Error")

    print("Test MaxHeap completed.")

if __name__ == "__main__":
    test_max_heap()

六、Heapify和Replace

1. Replace

什么是Replace？

取出最大元素后，放入一个新元素。

怎么实现？

可以先执行extract_max操作，再进行一次add操作来实现，但是这样的效果不是很好，因为这样的话需要执行两次O(log n)的操作；所以我们使用其他的实现方式：可以将堆顶的元素替换以后Sift Down，这样的话我们只需要执行一次O(log n)操作。

def replace(self, e: T) -> T:
    """取出堆中的最大元素，并替换成元素e"""
    ret = self.find_max()

    self._data.set(0, e)
    self._sift_down(0)

    return ret

2. Heapify

什么是Heapify？

将任意数组整理成堆的形状。

怎么实现？

遍历一下数组，然后将数组中的元素按照二叉堆的特性进行排序即可，但是由于这样做效率会比较低这里并不推荐。

更高效的实现

如图所示这是一个完全二叉树，但是并不满足二叉堆的特性。

所以我们根据上图中的二叉树找到倒数第一个非叶子节点，然后对这个节点按照倒叙从后往前不断的进行Sift Down的操作就可以了；但是这里有一个问题就是我们怎么确定倒数第一个非叶子节点的索引是多少呢？我们可以从最后一个节点的位置计算出最后一个非叶子节点的索引。

实现过程请参考图示：

至此我们的Heapify的操作就完成了。

代码实现

def __init__(self, arr: List[T] = None, capacity: int = 10):
    """构造函数，可以传入数组进行heapify"""
    if arr is not None:
        self._data = Array[T](len(arr))
        for element in arr:
            self._data.add_last(element)
        # Heapify操作
        for i in range(self._parent(len(arr) - 1), -1, -1):
            self._sift_down(i)
    else:
        self._data = Array[T](capacity)

测试

import time

def test_heap(test_data: List[int], is_heapify: bool) -> float:
    start_time = time.time()

    if is_heapify:
        max_heap = MaxHeap(test_data)
    else:
        max_heap = MaxHeap[int]()
        for num in test_data:
            max_heap.add(num)

    arr = []
    for _ in range(len(test_data)):
        arr.append(max_heap.extract_max())

    for i in range(1, len(test_data)):
        if arr[i-1] < arr[i]:
            raise ValueError("Error")

    print("Test MaxHeap completed.")

    end_time = time.time()
    return end_time - start_time

def main():
    n = 1000000

    test_data = [random.randint(0, 2**31 - 1) for _ in range(n)]

    time1 = test_heap(test_data, False)
    print(f"Without heapify: {time1} s")

    time2 = test_heap(test_data, True)
    print(f"With heapify: {time2} s")

if __name__ == "__main__":
    main()

七、基于堆的优先队列

1. 实现优先队列接口

from abc import ABC, abstractmethod

class Queue(ABC, Generic[T]):
    """队列接口"""

    @abstractmethod
    def get_size(self) -> int:
        """获取队列中元素的个数"""
        pass

    @abstractmethod
    def is_empty(self) -> bool:
        """判断队列中是否是空的"""
        pass

    @abstractmethod
    def enqueue(self, e: T) -> None:
        """向队列中添加一个元素(入队)"""
        pass

    @abstractmethod
    def dequeue(self) -> T:
        """从队列中删除一个元素(出队)"""
        pass

    @abstractmethod
    def get_front(self) -> T:
        """获取队首的元素"""
        pass