【Redis】跳跃表原理分析与基本代码实现(java)

最近开始看Redis设计原理，碰到一个从未遇见的数据结构：跳跃表(skiplist)。于是花时间学习了跳表的原理，并用java对其实现。

介绍

跳跃表是一种有序数据结构，它通过每个结点中维持多个指向其它结点的指针，从而达到快速访问结点的目的。

我们平时熟知的链表，查找效率为O(N)。跳表在链表的基础上，每个结点中维护了很多指向其它结点的指针，大大缩短时间复杂度。可以实现时间复杂度平均O(logN)，最坏O(N)。后文会有具体的分析和计算。

一个跳跃表示意图：

由左至右依次是，跳跃表结构结点(存储跳表信息)、头结点、连续的跳表结点。

最外层的跳表字段结构如下所示：

public class SkipList<T extends Comparable<? super T>> {

    //首尾结点的指针
    private SkipListNode<T> header;
    private SkipListNode<T> tail;

    //记录跳表中结点数量
    private long length;

    //最大结点的层数
    private int level;
    
    //...
}

跳表节点

跳表节点记为SkipListNode，内部字段结构如下：

class SkipListNode <T> {

    //索引层
    private SkipListLevel[] level;

    //后退指针
    private SkipListNode<T> backword;

    //分值
    private double score;

    //成员对象
    private T obj;
    
    //......
}

• 索引层数组：多个索引层组成的数组，每个元素包含一个指向其它节点的指针。通过这些指针的访问来加快查找速度。
• 后退指针：指向前一个节点；
• 分值：是一个浮点数，跳表中所有节点都按照分值从小到大来排序；
• 成员对象：即指向具体的数据对象。

索引层

索引层SkipListLevel的结构如下：

class SkipListLevel{

    //前进指针
    private SkipListNode forward;

    //跨度
    private int span;
    
    //......
}

• 前进指针：指向后续节点；
• 跨度：与指向的节点之间的距离。譬如，相邻节点距离就是1。

到这里，我们对跳表的基本结构有了一个清晰的认识。

理想的跳表

这里想先讲讲理想状态的跳表，不然无法理解实际跳表为什么可以缩减时间复杂度。

跳表节点间的关联方式：（索引层中的前向指针）第一层逐个链接，第二层每隔t个节点进行链接，第三层每隔2*t个节点进行链接，不断迭代。这里取t=2，画出每个节点的索引层之间的关联关系，得到如下图形式的链式结构：

有点像完全二叉树的结构。因此很容易理解：节点总数为N时，层最大高度为1+logN。例如图中有8个节点，最大层高为4。

搜索规则：从头结点的索引层的末端开始向下遍历。如果第K层的下一节点小于target，则移到该节点；若不小于，则下移到第K-1层。

按照此搜索规则，假设需要查找的target为7a，则搜索路径为0d--8d--0c--4c--4b--6b--6a--7a，如下图所示：

上述过程中，分别在8d、4c、6b、7a处进行比较。可见每一层都比较了一次，所以比较次数等于层数，为logN+1。所以时间复杂度为O(logN)。

如果实际的跳表按照这种形式进行设计，每次插入节点时，需要对很多结点的索引层进行调整，节点的插入删除将成为极其复杂的工作。因此，实际的跳表使用一种基于概率统计的算法，简化插入删除带来的调整工作，同时也能得到O(logN)的时间复杂度。

实际的跳表

每当需要新增一个节点时，需要考虑如何确定该节点的索引层层数，即SkipListLevel[]数组的长度。

如何确定“层”的高度？

在redis中，每次创建一个节点，都会根据幂次定律随机生成一个介于1和32之间的值作为索引层的高度。问题是，这个随机的过程如何设计？

我们观察理想状态跳表，可以发现，不算头节点总共8个节点，其中4个节点拥有2层索引，2个节点拥有3层索引，1个节点拥有4层索引。

可以近似看作满足这样的规律：节点索引层高度为 j 的概率为 1/2^j。因此每次生成新节点时，通过这样的概率计算可以得到索引层层数。代码如下所示：

/**
 * 获取随机的层高度
 * @return
 */
private int getRandomHeight() {
    Random random = new Random();
    int i = 1;
    for (; i < 32; ++i) {
        if (random.nextInt(2) == 0) {
            break;
        }
    }
    return i;
}

注意：在redis中最大索引高度不超过32

为什么时间复杂度平均O(logN)，最坏O(N)？

当节点数量足够多时，这种方式得到的跳跃表形态可以逼近理想的跳表的。很惭愧我不知道怎么证明，学过概率统计的同学一定很容易理解。它的时间复杂度就是近似为 O(logN) 。当然也有不理想的情况，当跳表中每一个节点随机得到的层高度都是 1 时，跳表就是一个普通双向链表，时间复杂度为 O(N) 。因此，时间复杂度平均O(logN)、最坏O(N)，这种说法是比较严谨的。

节点的分值

这个分值 score 很容易与节点的“跨度”混淆。跨度其实就是节点在跳表中的排位，或者说序号。而分值是一个节点属性。节点按照分值大小由小到大排列，不同节点的分值可以相等。如果分值相等，对象较大的会排在后面(靠近表尾方向)。

在实际API应用中，需要以分值和obj成员对象作为target进行查询、插入等操作。

功能实现

跳跃表的初始化-代码实现

class SkipList：

//构造方法初始化SkipList
public SkipList() {
    SkipListNode<T> node = new SkipListNode<>(null);
    this.header = node;
    this.tail = node;
    this.length = 0;
    this.maxLevelHeight = 0;
}

class SkipListNode：

//初始化头结点
SkipListNode(T obj){
    this.obj = obj;
    this.level = new SkipListLevel[32];
    initLevel(this.level,32);
    this.score = 0;
}
//根据"层高"和"分值"，新建一个节点
SkipListNode(T obj, int levelHeight,double score){
    this.obj = obj;
    this.level = new SkipListLevel[levelHeight];
    initLevel(this.level,levelHeight);
    this.score = score;
}
private void initLevel(SkipListLevel[] level, int height){
    for(int i=0;i<height;++i){
        level[i] = new SkipListLevel();
    }
}

跳跃表的插入-代码实现

流程如下：

• 按照幂次定律获取随机数，作为索引层的高度levelHeight，实例化新节点target；
• 设置一个SkipListNode类型的数组，update[]（记录所有需要进行调整的前置位节点，包括需要调整forword、或者只需要修改span值的节点），update[]的大小为max(levelHeight,maxLevelHeight);
• 设置int数组rank[]，记录update[]数组中各个对应节点的排位
• 遍历 update[] 进行插入和更新操作；根据update[]获取插入位置节点，进行插入；根据rank[]来辅助更新跨度值span。

实际代码比上述流程要复杂很多，levelHeight与maxLevelHeight的大小关系不能确定，根据不同的情况要对update[]进行不同的处理。

跳跃表插入的代码如下所示：

注意：是依据score大小和obj的大小来决定插入顺序

public SkipListNode slInsert(double score, T obj) {
    int levelHeight = getRandomHeight();
    SkipListNode<T> target = new SkipListNode<>(obj, levelHeight, score);
    // update[i] 记录所有需要进行调整的前置位节点
    SkipListNode[] update = new SkipListNode[Math.max(levelHeight, maxLevel)];
    int[] rank = new int[update.length];//记录每一个update节点的排位
    int i = update.length - 1;
    if (levelHeight > maxLevel) {
        for (; i >= maxLevel; --i) {
            update[i] = header;
            rank[i] = 0;
        }
        maxLevel = levelHeight;
    }
    for (; i >= 0; --i) {

        SkipListNode<T> node = header;
        SkipListNode<T> next = node.getLevel()[i].getForward();
        rank[i] = 0;
        //遍历得到与target最接近的节点(左侧)
        while (next != null && (score > next.getScore() || score == next.getScore() && next.getObj().compareTo(obj) < 0)) {
            rank[i] += node.getLevel()[i].getSpan();
            node = next;
            next = node.getLevel()[i].getForward();

        }
        update[i] = node;
    }

    //当maxLevel>levelHeight，前面部分节点的span值加1，因为该节点与forword指向节点之间将要 多出来一个新节点
    for (i = update.length - 1; i >= levelHeight; --i) {
        int span = update[i].getLevel()[i].getSpan();
        update[i].getLevel()[i].setSpan(++span);
    }
    //遍历 update[] 进行插入和更新操作
    for (; i >= 0; --i) {

        SkipListLevel pre = update[i].getLevel()[i];
        //将target节点插入update[i]和temp之间
        SkipListNode<T> temp = pre.getForward();
        int span = pre.getSpan();

        pre.setForward(target);
        pre.setSpan(rank[0] + 1 - rank[i]);

        target.getLevel()[i].setSpan(span > 0 ? (span - rank[0] + rank[i]) : 0);
        target.getLevel()[i].setForward(temp);
        //设置后退指针
        if (temp == null) {
            target.setBackword(header);
        } else {
            target.setBackword(temp.getBackword());
            temp.setBackword(target);
        }

    }

    if (tail.getLevel()[0].getForward() != null) {
        tail = target;
    }
    length++;
    return target;

}

跳跃表的节点删除-代码实现

根据分值和成员对象来删除跳表中对应节点

/**
 * 删除节点
 * @param obj
 * @return 删除的节点(若节点不存在则返回null)
 */
public SkipListNode zslDelete(double score, T obj) {
    SkipListNode[] update = new SkipListNode[maxLevelHeight];
    SkipListNode<T> node = header;
    for (int i = maxLevelHeight - 1; i >= 0; --i) {
        SkipListNode<T> next = node.getLevel()[i].getForward();
        //遍历得到与target最接近的节点
        while (next != null && (score > next.getScore() || score == next.getScore() && next.getObj().compareTo(obj) < 0)) {
            node = next;
            next = node.getLevel()[i].getForward();
        }
        update[i] = node;
    }
    //待删除的目标节点
    SkipListNode<T> target = update[0].getLevel()[0].getForward();
    if(target==null) return null;

    for (int i = maxLevelHeight - 1; i >= 0; --i) {
        SkipListLevel current = update[i].getLevel()[i];
        SkipListNode<T> next = current.getForward();
        if (next == null) continue;
        if (next != target) {
            current.modifySpan(-1);
            continue;
        }
        current.setForward(target.getLevel()[i].getForward());
        if(current.getForward()!=null)
            current.modifySpan(target.getLevel()[i].getSpan() - 1);
        else
            current.setSpan(0);
    }
    length--;
    while(header.getLevel()[maxLevelHeight-1].getSpan()==0){
        maxLevelHeight--;
    }
    return target;
}

跳跃表的节点查询-代码实现

1. 根据分值范围 fromScore~toScore，返回第一个符合范围的节点

• 参数 node 是开始查询的位置，调用时传入header ， 递归过程会发生变化；
• k 是当前层数，从最高层开始递归遍历；

public SkipListNode<T> zslFirstInRange(double fromScore, double toScore, SkipListNode<T> node, int k) {
    if (!zslIsInRange(fromScore, toScore)) {
        return null;
    }

    SkipListNode<T> next = node.getLevel()[k].getForward();

    if (next == null || next.getScore() >= fromScore) {
        if (k == 0) return next != null && next.getScore() > toScore ? null : next;
        return zslFirstInRange(fromScore, toScore, node, k - 1);
    }
    return zslFirstInRange(fromScore, toScore, next, k);
}

2. 根据分值范围，返回最后一个符合范围的节点

public SkipListNode<T> zslLastInRange(double fromScore, double toScore, SkipListNode<T> node, int k) {
    if (!zslIsInRange(fromScore, toScore)) {
        return null;
    }

    SkipListNode<T> next = node.getLevel()[k].getForward();

    if (next == null || next.getScore() > toScore) {
        if (k == 0) return next != null && next.getScore() < fromScore ? null : node;
        return zslLastInRange(fromScore, toScore, node, k - 1);
    }
    return zslLastInRange(fromScore, toScore, next, k);
}

本篇博客介绍了跳跃表基本原理，并使用java完成了基本数据结构的封装，实现了节点“插入”、“删除”、“搜索”等核心功能的代码实现。