[List] 手撕跳表

Git 实现代码

跳表实现总结。

参考《Redis 设计与实现》、《Skip Lists, A Probabilistic Alternative to Balanced Trees》

定义

为了尽可能讲明白跳表,先要提一个熟悉的数据结构:链表。

即,我们如何从链表引申出去,最后设计出跳表的。

假设有如下结构的链表:

我们都知道,如果要找 5,必须从头开始遍历链表,直到找到 5。

为了更快的找到 5,我们可以给这个单链表增加多层索引,如下:

上图中,索引设定的规则是二分,每向上一层,指针每次前进都会跳过一倍数量的节点。

但是这个结构存在一个重大的问题:如何在添加删除节点后,低成本的维护索引的平衡?

我们是否有必要严格维护索引构建的规则呢?


跳表与 B+树 最大的不同,在于使用概率来控制分布平衡,这也是跳表最重要的核心思路。

这里先给出节点层级的生成代码来引入,如下:

private int randomLevel() {
    int level = 1;
    Random random = new Random();
    for (int i = 1; random.nextInt(2) == 1 && i < MAX_LEVEL; i++) {
        level++;
    }
    return level;
}

可以看到,上述代码中生成的 level 每要增加 1,就得经过一个 50% 概率的筛选(此概率可调控)。那么假设调用此方法生成 2^10 个 level 值(1024 次调用),概率上会出现 1 个 11,2 个 10,4 个 9,8 个 8,16 个 7 ……

即 2^(n-1) 次调用,会出现 2^0 次 n,2^1 次 (n-1),2^2 次 (n-2),2^3 次 (n-3) …… 2^k 次 (n-k) (n > k > 1)

不同层高的节点数量,和二分地给链表添加多层索引一样。

跳表表明,没有必要严格维护一个索引来达到平衡,可以通过概率来实现绝大多数情况下的平衡。规则更为宽松,但效果不差。如下为论文中的示例:

结构

很多人认识跳表可能都是从 Redis 入手。Redis 中对论文提出的跳表进行了实现和优化,我参照 Redis 的属性设计,但方法可能会和 Redis 的方法实现有较大差异,如下:

public class SkipList<T> {
    // 指向首节点,首节点有 MAX_LEVEL 层索引,key value 皆为空
    // 可以理解为一个能到全表所有位置的起点
    private SkipListNode<T> header;

    // 尾节点
    private SkipListNode<T> tail;

    // 最大节点的高度(不算首节点)
    private Integer level;

    // 总节点数(不算首节点)
    private Integer length;

    // 最大节点层数
    private static final Integer MAX_LEVEL = 32;

    class SkipListNode<T>{}
    class SkipListLevel<T>{}

    T find(String key);
    void insert(String key, T value);
}
class SkipListNode<T> {
    // SkipListLevel 集合
    protected SkipListLevel<T>[] levels;

    // 指向上一个节点的指针
    protected SkipListNode<T> backward;

    // key 和 value 没什么说的
    // Redis 中对应为 score 和 obj
    protected String key;
    protected T value;

    T find(String key, Integer curLevel);
    SkipListNode<T> findPos(String key, Integer curLevel, Integer random, SkipListNode insertNode);
}
class SkipListLevel<T> {
    // 当前层指向下一个节点的指针
    protected SkipListNode<T> forward;
    // span 跨度,仅用于查找排位,与 score 并无关系
    // 表明当前节点到本层中 forward 指向的下个节点要经过几个最底层节点
    // (包括当前节点,所以最小为 1)
    //protected Integer span;
}

操作

插入和查询是密不可分的,因为插入的前提是查询到插入位置。

查询

上图也是从论文中截取,我们可以看到,我们通过首节点中,当前跳表中节点存在的最高层数作为起始层,开始查询(如果起始层比当前节点最高层数还高,那肯定会直接指向 NIL)。

首先比对要查询的 key 和 当前层指向的节点中的 key 孰大孰小。如果 查询的 key 更大,说明可以跳过去,如果要查询的 key 更小,则向下移动一层,继续比对。

注意这个过程中,层数一定是逐渐下降的,不可能出现上升的情况,所以可以提出层数做个递归。

class SkipListNode<T> {

    ......

    // 查询值
    private T find(String key, Integer curLevel) {
        for (;curLevel >= 0; curLevel--) {
            // 跳过指向 null 的 level.forward
            if (levels[curLevel].forward == null) {
                continue;
            }
            if (key.compareTo(levels[curLevel].forward.key) == 0) {
                return levels[curLevel].forward.value;
            }
            if (key.compareTo(levels[curLevel].forward.key) > 0) {
                return levels[curLevel].forward.find(key, curLevel);
            }
        }
        return null;
    }
}

插入

继续看上图,在插入前我们需要知道插入在哪个节点后。但是如果我们先找到那个节点,再随机生成高度,创建新节点的话,如果这个生成的新节点很高,我们就需要倒回前面的节点,去修改他们 levels 中 level 的指向。

举个例子,你可以想象在上图中椭圆位置,随机生成了一个新的最高的节点来插入,那意味着 key 为 6 的节点的最上两层 Level,key 为 9 的节点最上一层 Level 都需要修改 forward 指向新插入的节点。

为了避免这种情况产生,可以选择预先随机生成好高度并创建好新节点,在查找插入节点位置的过程中修改指针指向为新节点,因为我们提前知道了新节点的高度,那么在查找过程中高度不大于新节点高度的层,如果指向的节点 key 比新节点的 key 要大,意味着会被新节点拦截,就直接修改指针指向(见 findPos() 方法)。

PS:Redis 中添加的 span 也可以在这个过程中维护。

class SkipListNode<T> {

    ......

    // 找到插入节点位置
    private SkipListNode<T> findPos(String key, Integer curLevel, Integer random, SkipListNode insertNode) {
        for (;curLevel >= 0; curLevel--) {
            // 跳过指向 null 的 level.forward
            if (levels[curLevel].forward == null) {
                if (curLevel < random && curLevel > 0) {
                    insertNode.levels[curLevel].forward = levels[curLevel].forward;
                    levels[curLevel].forward = insertNode;
                }
                continue;
            }
            if (key.compareTo(levels[curLevel].forward.key) > 0) {
                return levels[curLevel].forward.findPos(key, curLevel, random, insertNode);
            }
            if (curLevel < random && curLevel > 0) {
                insertNode.levels[curLevel].forward = levels[curLevel].forward;
                levels[curLevel].forward = insertNode;
            }
        }
        return this;
    }
}
public class SkipList<T> {

    ......

    public void insert(String key, T value) {

        // 先计算出节点的随机层高,创建新节点
        int random = randomLevel();
        SkipListNode insertNode = new SkipListNode(key, value, random);
        SkipListNode base = header.findPos(key, header.levels.length - 1, random, insertNode);

        // 赋值是否是当前最大 level
        if (insertNode.levels.length > level) {
            level = insertNode.levels.length;
        }

        // 前一个节点不为头结点
        if (base != header) {
            insertNode.backward = base;
        }

        // 找到下一个节点
        SkipListNode nextNode = base.levels[0].forward;

        if (nextNode != null) {
            nextNode.backward = insertNode;
        } else {
            tail = insertNode;
        }
        insertNode.levels[0].forward = nextNode;
        base.levels[0].forward = insertNode;

        length++;
    }
}