二叉堆

二叉堆具有两个性质, 结构性和排序性.

结构性质

堆是一棵除了底层以外都被完全填满的二叉树, 叫做完全二叉树, 而最底层从左到右都是满的, 右侧允许有空缺

以上图为例, 1是小顶堆, 2是大顶堆, 3是小顶堆, 4不是堆.
这种完全二叉树从树的形状看并没有被填满, 为什么要被称作是完全二叉树呢? 这是因为这种二叉树可以被高效的保存在数组中. 二叉树通常有两种方式存储:

基于链表

每个节点存储3个字段

基于数组

这种方式中根节点存储在1位, 后面左节点存储在 2i位, 右子节点存储在2i+1位. 通过这种方式只要知道根节点的位置, 就可以利用上述的关系构建出整个树, 这样的存储非常的紧密, 不需要额外存储左右指针. 之所以根节点存在1的位置, 是因为根据父亲节点计算子节点或者子节点计算父节点都只需要 乘以 2 或者 除以 2 通过位运算可以高效实现.
但是如果二叉树不是满二叉树按照这种方式存储就会存在空洞, 造成空间浪费

回到堆的话题, 因为堆满足完全二叉树的结构性, 所以通常堆都是存储在数组中.

堆序性质

在一个堆(小根堆)中, 对于每一个节点X, X的父亲节点 <= X节点

如上图右侧的完全二叉树就不是堆.

从上面定义的信息可以看出堆所提供的有序性是有限的, 只能知道堆顶是最大或最小值, 因此堆最常用的就是用作优先级队列, 优先级队列所需要的接口.

• add
• findMin
• deleteMin

在添加和删除的时候都需要去维护上面提到的两个堆的性质.

实现

add

新插入的值默认插在末尾, 这会破快堆的有序性, 所以需要找到新加入的值合适的位置. 将插入的值逐步和parent比较, 如果小于parent 则上浮

样例实现

public void add(int x) {
    if (size == elements.length - 1) {
    enlargeArray();
}

// 复制elements[0]的好处是for循环不需要对 index 0 特殊处理
// i = 1时, 就会和elements[0] 作比较
elements[0] = x;
// percolate up / sift up
// i 要从 ++size算起, 不然找的parent可能是错的
for (int i = ++size; i >= 1; i = i / 2) {
    // 插入值小于父节点, 将父节点下移, 空穴上移
    if (x < elements[i / 2]) {
        elements[i] = elements[i / 2];
    } else {
        elements[i] = x;
        break;
    }
}
}

deleteMin

书上画的做法如下

1. 判断是否右子树为空或者左子树小于右子树, 如果是则左子树上移, 否则右子树上移
2. 退出条件为移动到没有子树的节点 即 i > size /2
3. 退出后将最后一位赋值给当前值

需要注意的是左右 子树不一定都存在

private void deleteMin() {
    int i = 1;
    if (size == 1) {
        size = 0;
        return;
    }
    for (; i <= size / 2;) {
        // 右子树为空或者左子树小于右子树
        if (2 * i + 1 > size || elements[2 * i] < elements[2 * i + 1]) {
            // 左子树上移
            elements[i] = elements[2 * i];
            i = 2 * i;
        } else {
            elements[i] = elements[2 * i + 1];
            i = 2 * i + 1;
        }
    }

    elements[i] = elements[size];
    size--;
}

但是他给出的代码实现却是另一种思路
首先将最后一位填充到根节点处, 同时这不一定满足堆序性质, 所以要和左右子树逐步比较下沉.

/**
 * 优化版本的deleteMin
 * 1. 先将末尾放到堆顶
 * 2. 从堆顶开始堆化处理
 */
private void deleteMin2() {
    if (size == 0) {
        throw new IllegalArgumentException("Empty!");
    }
    // 注意要先减1 在 siftDown, 因为重新堆化时总大小变了
    elements[1] = elements[size--];
    siftDown(1);
    // size--;
}

/**
 * 支持从某节点开始将其以下部分调整至满足堆的特性
 * @param start
 */
private void siftDown(int start) {
    int i = start;
    int child;
    int value = elements[start];
    for (; i <= size / 2; ) {
        child = i * 2;
        // 有右子树, 右子树更小, 那么下移一位
        if (child != size && elements[child + 1] < elements[child]) {
            child++;
        }

        // 如果和elements[i] 比较, 每次都要进行swap 需要3次赋值
        // 如果和value比较, 每次只需要一次赋值.
        // if (elements[child] < elements[i]) {
        if (elements[child] < value) {
            elements[i] = elements[child];
        } else {
            // 父节点比两个子树都小
            break;
        }
        i = child;
    }
    elements[i] = value;
}

这两种实现差别

• 第一种的思路是不断的在两个子树中查找较小值提升
• 第二种思路是将末尾设置为root后, 不断地将这个值下沉, 第二种更符合算法的 percolate down的描述. 而且第一种一定会比较logN次, 而第二种实际上是可能会提前终止的. 所以第二种的最优运行情况中的比较次数会更少

打印

同样添加了一个方法来打印堆的结构方便检验准确性

private void print() {
    int currentLevel = 0;
    int currentIndent = 0;
    int totalLevel = (int) Math.floor(Math.log(size) / Math.log(2)) + 1;
    StringBuilder levelBuilder = new StringBuilder();
    levelBuilder.append(String.format("level% 2d: ", currentLevel));
    for (int i = 1; i <= size; i++) {
        // 不精确的计算 log2(N)
        int level = (int) Math.floor(Math.log(i) / Math.log(2));
        if (level != currentLevel) {
            System.out.println(levelBuilder.toString());
            currentLevel = level;
            currentIndent = 0;
            levelBuilder = new StringBuilder();
            levelBuilder.append(String.format("level% 2d: ", currentLevel));
        }
        int toLeftIntent = indent(i, totalLevel - 1);
        levelBuilder.append(blank((toLeftIntent - currentIndent), currentLevel));
        currentIndent = toLeftIntent;
        levelBuilder.append(String.format("%d", elements[i]));
    }
    System.out.println(levelBuilder.toString());
}

堆的应用

第一个应用就是上面提到的优先级队列, 例如java里面的PriorityQueue

堆排序

如果基于原始数据构建一个堆, 并不断地提取堆顶的元素就可以达到排序的效果. 这里会分为两步, 建堆和排序

建堆

// 从 n/2 也就是所有的非叶子节点满足堆的性质就完成了堆化
for (int i = size / 2; i >= 1; i--) {
    siftDown(i);
}

可以原地完成建堆, 将size / 2 ~ 堆顶的数据循环进行siftDown, 也就是将所有的非叶子节点调整满足堆的性质, 整体堆性质就满足了, 这样后面那一半的叶子节点的是不用遍历的, 减少了一半的循环比较. 这里忽略了数组第0位, 实际如果原地数组排序还是需要考虑这一点的

建堆的复杂度分析
因为叶子节点不需要堆化，所以需要堆化的节点从倒数第二层开始。每个节点堆化的过程中，需要比较和交换的节点个数，跟这个节点的高度 k 成正比。所以我们需要求所有非叶子节点的高度和.

高度 h 一层有1个节点,
高度 h-1 一层有\(2^1\)节点,
高度 h-2 一层有 \(2^2\)节点
高度 h - i 一层有 \(2^i\)节点
所以对于高度h的堆, 所有节点的高度和是多少?
\(S = \sum_{i=1}^{h}{2^i(h-i)}\)
\(S = h + 2(h - 1) + 4(h - 2) + ... + 2^{h-1}(1)\)
\(2S = 2h + 4(h - 1) + 8(h - 2) + ... + 2^h(1)\)
两式相减得到
\(S = -h + 2 + 4 + 8 + ... + 2^{h - 1} + 2^h\)
\(S = (2^{h+1} - 1) - (h + 1)\)
因为 h = \(log_2N\)所以 可以得到 S 即建堆的复杂度是 O(N).

进一步分析 因为N介于 \(2^h\)到 \(2^{h + 1}\)之间 完全二叉树到完美二叉树.
所以 S 的范围实际上是 N ~ 2N的复杂度, 当N = 2^h 时, 复杂度就接近2N. 所以堆排序并不优, 弱于快速排序

排序

private int[] sort() {
    for (int i = size; i >= 1; i--) {
        int v = deleteMin();
        // 将堆顶元素放到最后一位 
        elements[i] = v;
    }
    return elements;
}

堆排序整体时间复杂度:** 建堆过程 O(N)**, 排序过程是O(NlogN) 所以整体式O(nlogN)

TopK

例如要求前三大的数字. 那么可以维护一个小顶堆, 每来一条数据, 如果比堆顶的数据小, 则不动, 如果比堆顶的数据大则移除堆顶的数据, 插入新数据. 小顶堆的作用是能将来的数据迅速和当前的最小值比较, 小于则直接淘汰
只要在add的逻辑中稍加修改即可支持topK计算

if (size >= maxSize) {
    if (x > findMin()) {
        deleteMin();
    } else {
        return;
    }
}

定时器

例如Flink中的timer service. 通过维护小根堆, 当watermark推进时, 只需要不断地从堆顶poll最小值来判断是否小于当前的watermark, 如果小于则需要触发计算.
参看 org.apache.flink.runtime.state.heap.HeapPriorityQueue
添加数据

@Override
protected void addInternal(@Nonnull T element) {
    final int newSize = increaseSizeByOne();
    moveElementToIdx(element, newSize);
    siftUp(newSize);
}

1. 调整数组大小
2. 将新元素至于下一index
3. siftUp 直至满足堆性质

private void siftUp(int idx) {
    final T[] heap = this.queue;
    final T currentElement = heap[idx];
	// 使用位移加速热点代码计算
    int parentIdx = idx >>> 1;

    while (parentIdx > 0 && isElementPriorityLessThen(currentElement, heap[parentIdx])) {
        moveElementToIdx(heap[parentIdx], idx);
        idx = parentIdx;
        parentIdx >>>= 1;
    }

    moveElementToIdx(currentElement, idx);
}

而删除他不仅支持了deleteMin, 还支持任意位置的删除. 为啥要支持这个呢, 因为在flink中可以支持删除任意的已经注册过的timer, 所以需要支持任意删除的功能. 为了支持任意删除的功能, 首先在插入进去的元素上记录了存储的下标.
每次移动到数组中某个位置时, 就会更新这个element内部的index.

protected void moveElementToIdx(T element, int idx) {
    queue[idx] = element;
    element.setInternalIndex(idx);
}

为了实现这一点, 在HeapPriorityQueue之上封装了HeapPriorityQueueSet提供去重语义(包括在add时如果是已经添加过的某个key的某个时间戳的timer, 也会通过此来去重不再添加). 去重就是通过hashMap来维护的

@Override
public boolean add(@Nonnull T element) {
    return getDedupMapForElement(element).putIfAbsent(element, element) == null
            && super.add(element);
}

@Override
public boolean remove(@Nonnull T toRemove) {
    T storedElement = getDedupMapForElement(toRemove).remove(toRemove);
    return storedElement != null && super.remove(storedElement);
}

在remove时, 是通过用户指定的timestamp 和 key 去 Map中查找到已经记录了位置的 HeapPriorityQueueElement.

public boolean remove(@Nonnull T toRemove) {
    final int elementIndex = toRemove.getInternalIndex();
    removeInternal(elementIndex);
    return elementIndex == getHeadElementIndex();
}

protected T removeInternal(int removeIdx) {
    T[] heap = this.queue;
    T removedValue = heap[removeIdx];

    assert removedValue.getInternalIndex() == removeIdx;

    final int oldSize = size;

    if (removeIdx != oldSize) {
        T element = heap[oldSize];
        moveElementToIdx(element, removeIdx);
        adjustElementAtIndex(element, removeIdx);
    }

    heap[oldSize] = null;

    --size;
    return removedValue;
}

private void adjustElementAtIndex(T element, int index) {
    siftDown(index);
    if (queue[index] == element) {
        siftUp(index);
    }
}

1. 首先从元素中取得index
2. 删除的做法和删除堆顶类似, 将末尾填充到删除的位置, 然后执行siftDown, siftDown之后如果发现index处的位置没有变, 那么说明此元素比两个子节点都小, 就尝试siftUp

这种应用有点类似于书里提到的延伸应用, 因为堆内部的数据的信息有限, 所以如果有额外的数据结构快速定位到某个数据, 可以实现定点删除, 调整优先级等策略, 思路无外乎找到元素, 删除或者调整数值, 重新siftDown或siftUp.

多路合并

归并排序中实际上是两路合并, 如果有多个有序序列, 要合并成一个全局有序的序列, 就可以借助堆来计算.

例如leetcode Q23: 合并 K 个升序链表, 利用堆来辅助可以很轻松的完成

public ListNode mergeKLists(ListNode[] lists) {
    if (lists.length == 0) {
        return null;
    }
    PriorityQueue<ListNode> queue =
            new PriorityQueue<>(
                    lists.length,
                    new Comparator<ListNode>() {
                        @Override
                        public int compare(ListNode o1, ListNode o2) {
                            return Integer.compare(o1.val, o2.val);
                        }
                    });
    for (ListNode node : lists) {
        if (node != null) {
            queue.add(node);
        }
    }
    ListNode head = new ListNode();
    ListNode node = head;
    while (!queue.isEmpty()) {
        ListNode min = queue.poll();
        // 将val赋值给next节点, 这样防止了尾节点设置了初始值的问题
        node.next = new ListNode();
        node = node.next;
        node.val = min.val;
        if (min.next != null) {
            min = min.next;
            queue.add(min);
        }
    }
    return head.next;
}

小结:
二叉堆的数据结构虽然很简单, 但是应用场景相当的广泛.

参考

<数据结构与算法分析 Java描述> 6.3 BinaryHeap
极客时间 数据结构与算法之美 28
<Hbase原理与实践>
完全二叉树看起来并不完全，为什么叫完全二叉树呢？
https://blog.csdn.net/qq_42006733/article/details/104580717