渐进式解析 Redis 源码 - 跳跃表 skiplist

2018-11-20 阅读量

跳跃表 ( skiplist ) 为一种随机化的数据结构，以有序的方式在层次化的链表中保存元素，通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问的目的。

跳跃表在 Redis 主要是作为有序集合ZSET 类型数据的底层实现之一，如果有序集合里面包含的数量比较多，或者成员元素的键比较长（必须大于 zset-max-ziplist-value 设置的值默认 64，否则使用 ziplist ）；

假如有个至少数十万的数据需要进行按顺序存放，并且能按照最低复杂度增删改查。一般情况下都是使用线性的链表和数组实现，但是无论数组还是链表在写入的时候都是一个效率很低的操作：遍历获取存入位置，最快的方法就是使用二分查找，链表不能使用二分查找，但是呢，数组又需要插入位置之后的所有元素后移；总结来说，数组的插入复杂度是 O(n)，链表的查找复杂度是 O(n)，所以使用普通的线性表是一个恐怖的操作。

so，这个时候跳跃表 ( skiplist ) 就出现了…

跳跃表 ( skiplist ) 在增删改查的复杂度上平均能达到 O(log n)，最坏O(n)，至于为什么 Redis 使用了 skiplist 而不是各种比如红黑树的平衡树呢，且看下面详解 < 挖坑跑 >~~~

跳跃表对比于平衡表如下：

在做范围查找的时候，平衡树要比 skiplist 复杂得多，在平衡树中，我们找到指定范围的小值以后，还需要中序遍历顺序查找其他其他不超过大值得节点，这样的话就还需要对平衡树进行改造，否则这种中序遍历并不容易实现；但是在 skiplist 中就比较简单了，找到小值的时候直接继续挨着遍历就行了
如果进行增删操作的时候，平衡树可能会引起 rebalance 将子树的重新排列，以保持平衡；但是 skiplist 就更加简单了，只需要修改相邻节点指针即可，相当便利
一定情况下 skiplist 的内存占用会更有优势一些，Redis 里，平均一个节点包含 1.33个指针，而平衡树需要 2个指针
最后就是最重要的一点，skiplist 在实现上简单很多很多

跳跃表的推导

由于数组的特性，几乎无法避免写入时候的元素大面积移动，所以我们从链表入手，主要就是解决链表元素的快速定位问题，假如有一组有序数字（-INF、-10、3、10、21、36、43、49、59）组成的有序链表：

单链表

因为链表无法使用二分法查找，我们要查找到 59 这个值，不包括 -INF 负无穷大，需要 8 步，那么如何用更少的路径去查找到目的值呢？

跳跃表1

所以我们开辟一条路径试试，从 L2 来找的话，只需要 4步就能找到 59 这个值；获取 49 我们只需要查找 5次，因为找到 43 的时候还不到，而到 59 的时候显然已经超过了，所以往回找与 43之间的下一层（L1），很幸运我们只需要走 1步；而如果我们还需要再加快查找 59呢，根据上面的经验，我们可以另外再增加一层

跳跃表2

这样我们通过 L3 只需要查找两次即可，那么按照这个思路我们最终的结构就是这样了

跳跃表3

如果查找 59 我们只需要一次查找，查找 36 需要 5次，即 L4（-INF -> 59），L3（-INF -> 21、21 -> 59）、L2（21 -> 43）、L1（21 -> 36），由于数量比较少可能看不出来，如果数量上十万百万效率就比较清楚了，我们进行推导一下：

如果有 n 个元素，因为是 2 分，所以层数就应该是 log n 层（本文所有 log 都是以 2 为底），再加上自身的 1 层。以上图为例，如果是4个元素，那么分层为 L3和 L4，再加上本身的 L2，一共 3 层；如果是 8 个元素，那么就是 3 + 1 层。最耗时间的查询自然是访问所有层数，耗时 logn + logn，即 2logn。为什么是2倍的logn呢？我们以上图中的 49 为例，查询到 49 要访问所有的分层，每个分层都要访问 2 个元素，中间元素和最后一个元素。所以时间复杂度为 O(log n)

当然，我们上述为一种比较理想的跳跃表，而实际上的跳跃表上级跳跃点（L2、L3、L4）的点都是随机的，如果链表中又新增了一个节点，那么将采用 抛硬币 的方法来判定是否提拔这个节点为跳跃点，这里我们后续在代码的逻辑中也会说到

Redis 跳跃表结构

Redis 跳跃表的数据结构如下：

/* server.h: line 760 */
// 跳跃表 节点 结构
typedef struct zskiplistNode {
    // 成员对象 key 值结构，简单字符串类型
    sds ele;
    // 分值
    double score;
    // 后退指针，指向前一个节点
    struct zskiplistNode *backward;
    // level 层数组
    struct zskiplistLevel {
        // 前进指针，下一个跳跃点地址
        struct zskiplistNode *forward;
        // 跨度
        unsigned int span;
    } level[];
} zskiplistNode;
// 跳跃表 结构
typedef struct zskiplist {
    // 首 尾 节点的地址
    struct zskiplistNode *header, *tail;
    // 整个表 节点数量
    unsigned long length;
    // 最大层级
    int level;
} zskiplist;
// zset 结构
typedef struct zset {
    // 数据字典
    dict *dict;
    // 跳跃表数据
    zskiplist *zsl;
} zset;

假如下面有一个 ZSET 类型数据使用了跳跃表（假定符合跳跃表的条件…）

127.0.0.1:6379> ZRANGE fprice 0 100 withscores
 1) "banana"
 2) "1.5"
 3) "cherry"
 4) "2.3"
 5) "grape"
 6) "3.2"
 7) "peach"
 8) "4.2"

那么这个数据的图形结构如下（这里需要注意 实际程序中 level 是从 0 开始的，图例中从 1 开始仅仅是为了方便查看）：
跳跃表图形结构

zskiplist 部分

header 指向跳跃表表头地址
tail 指向跳跃表表尾地址
length 表示跳跃表长度，也是跳跃表目前包含节点的数量（不包括表头节点）
level 最大层级（不包括表头节点的层级）

zskiplistNode 部分，又分为表头节点和普通数据节点

表头节点 主要包含所有层级指针地址，默认最高 32层（所以集合中最大的成员数为 2³² - 1 = 4294967295，每个集合可存储40多亿个成员）
ele 和 score 分别表示成员对象和分值；成员对象为简单字符串 sds 类型数据，分值存储的为 double 浮点数，跳跃表里面的所有节点都是根据分值从小到大排序
backward 指向前一个节点地址，用于从表尾向表头访问节点，不过只能一次遍历一个，不像 level 里面的跳级访问后续节点
level 层级
- skiplistNode 里面的 level 数组属性包含多个 zskiplistLevel 类型的层级元素，每个元素都包含一个指向下一个节点的指针，程序可以通过这些层快速访问其他节点，一般层数越多，访问其他节点的速度就越快；每次创建一个新的数据节点，程序都根据幂次定律（power law，越大的数出现的概率越小）随机生成一个介于 1 ~ 32 的值作为 level 数组的大小，就是 level 层级数
- forward 每一层都有一个指向下一个节点的指针地址，用于从表头到表尾遍历所有节点，遍历到指向节点是 null 的时候表示到结尾
- span 跨度，用于记录两个节点之间的跨度值，两个节点的跨度越大，代表距离越远；当 forward 指向 NULL 的时候，span 为 0；乍看跨度是用来快速遍历的，实际上快速遍历只需要用 forward 的，实际上跨度是用来表示顺序rank 的，从开始到目标节点的跨度之和其实就是顺序值

下面我们就 Redis 跳跃表一些典型的 API 进行解析~~~ 不过为了表示效率，我们还会带上相关 API 的复杂度数据

源码解析

由于有些方法里的代码量比较大，我们这里按照典型的代码片段进行解析，同志们可以根据文章提示的代码位置和代码里面的关键词在源码中搜素，可能数据结构一些元素看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

跳跃表创建函数 zslCreate() 和 zslCreateNode()，逻辑比较简单，主要是申请内存空间、初始化属性；时间复杂度为 O(1)

跳跃表插入函数 zslInsert()，具体逻辑为：查找插入点、插入节点、更新前后节点的 forward、backward 属性、更新跳跃表属性等；不过主要的难点还是查找该在何处插入节点，跳跃表的顺序是按照 score 分值进行排列的，一般查找是从最高层 level 开始遍历查找，如果当前节点的 score 小于新 score 则继续往 forward 方向查找，否则，则降低一层继续往 forward 方向查找，直到找到小于 score 但是最接近的节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度

/* t_zset.c: line 132 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    // 检验 分值 score 的类型
    serverAssert(!isnan(score));
    // level 表头地址
    x = zsl->header;
    // 遍历跳跃表表头的 level所有层，从 最上层 往下遍历
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        // 初始化写入位置，也代表 写入位置距离当前距离
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 遍历当前 level 层，根据 forward 依次往后遍历，找出在当前层 插入位置 前方距离最近的 节点
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                    sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        // 最近的前方节点
        update[i] = x;
    }
    /* we assume the element is not already inside, since we allow duplicated
     * scores, reinserting the same element should never happen since the
     * caller of zslInsert() should test in the hash table if the element is
     * already inside or not. */
    // 一个 zset 里面不可能有相同 元素存在，因为已经使用 hash 判断过了，但是 可能存在 score 相同的元素
    // 给元素随机 level 层数，提拔的概率是随机的，当然肯定是 0 ~ 31 范围内
    level = zslRandomLevel();
    // 如果层级大于当前所有层级，则 需要从 header 表头进行添加 forward，而且要更新 span 跳跃距离
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }
    // 新建初始化的节点
    x = zslCreateNode(level,score,ele);
    // 根据刚才层级遍历查找的位置进行 新节点插入，而且要 重新设置 在每个 层级中 该节点的 forward 以及 跳跃距离span
    for (i = 0; i < level; i++) {
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;
        /* update span covered by update[i] as x is inserted here */
        // 更新插入点的 跳跃距离 span
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 更新插入点前一个节点的 跳跃距离 span
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    /* increment span for untouched levels */
    // 增加高层的每层中 插入点前一个元素的 跳跃距离span
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 设置后置 backward 元素地址
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    // 如果 新增的节点 level[0] 的 forward 存在，则设置 下一个元素的  后置地址backward 为新节点
    // 否则代表 新增的节点 为最后一个元素，则吧 跳跃表的 tail 地址指过来
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    // 新增跳跃表长度
    zsl->length++;
    return x;
}

跳跃表节点的删除分为三种情况

zslDelete() 删除指定成员、指定分值的节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度
zslDeleteRangeByScore() 删除指定分值区间范围内的所有节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度
zslDeleteRangeByRank() 删除指定排位范围内的所有节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度

然而这几个函数的核心删除节点函数都是 zslDeleteNode()，区别就是查找需要删除节点的逻辑不一样，我们这次以 zslDelete() 函数作为示例，不过这个函数实际查找节点的逻辑也是元素值ele 以及分值score 配合使用的

/* t_zset.c: line 218 */
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;
    // 从表头 header 进行遍历每层
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        // 使用 forward 遍历每层level，配合 score 进行区间查找，如果 score 相同而且 ele 相同证明已经找到了，反之，获取最近的高层节点，降level 查找的时候 开始节点为 刚才的高层节点
        while (x->level[i].forward &&
                (x->level[i].forward->score < score ||
                    (x->level[i].forward->score == score &&
                     sdscmp(x->level[i].forward->ele,ele) < 0)))
        {
            x = x->level[i].forward;
        }
        // 记录目标节点在每层的前一个节点地址，后续删除的时候需要使用
        update[i] = x;
    }
    /* We may have multiple elements with the same score, what we need
     * is to find the element with both the right score and object. */
    // 查找到的最近score 的节点，不过我们判断不止要判断 分值 score 是否符合，还需要判断 ele 是否一样
    x = x->level[0].forward;
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
        // 找到目标，执行删除 节点
        zslDeleteNode(zsl, x, update);
        // 这里的 node 主要用来判断是否释放这个节点，如果 node 传递为 NULL 则释放节点，否则就是在 更改指定元素的分值 score 的时候使用的，这里进行节点删除后不进行释放，而是移除节点后再进行返回，可以得以重用，特别是 ele这个 SDS 字符串结构
        if (!node)
            zslFreeNode(x);
        else
            *node = x;
        return 1;
    }
    return 0; /* not found */
}

删除跳跃表节点的时候，除了传递节点地址以外，一般还会传递这个节点在每层的前一个节点地址

/* t_zset.c: line 190 */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    // 遍历每个层次，配合前一个节点 进行 属性更新
    for (i = 0; i < zsl->level; i++) {
        // 如果 目标节点x 存在于当前 level 层，则变更 跳跃距离 span，以及 forward 地址
        // 否则该层的 跳跃距离 -1
        if (update[i]->level[i].forward == x) {
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            update[i]->level[i].span -= 1;
        }
    }
    // 如果下一个节点存在，则更新 回退地址 backward
    // 如果不存在，证明删除的为最后一个元素，则更新跳跃表 tail 属性
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    // 如果删除的是最高层节点，整个跳跃表 层级 -1
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    // 跳跃表长度 -1
    zsl->length--;
}

获取节点排位函数 zslGetRank() 和获取指定排位的节点函数 zslGetElementByRank()，这两个函数的逻辑实际比较简单，也比较类似，查找逻辑跟 zslDelete() 函数比较相像，不过区别在于利用了 rank 属性的值，我们在上面 zskiplistNode结构部分的介绍中，提到 rank 其实就是用来表示排名的，header 到目标节点之间每层 level 查找路径的 span 跨度其实就是排位；两个函数的时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度

ZSET 的区间操作API有两个：

zslFirstInRange() 获取在跳跃表中符合给定范围分值的第一个节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度
zslLastInRange() 获取在跳跃表中符合给定范围分值的最后一个节点；时间复杂度为 平均 O(log n)，最坏 O(n)，n 代表跳跃表长度

/* t_zset.c: line 274 */
zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;
    /* If everything is out of range, return early. */
    // 判断给定的区间是否符合 跳跃表 最大最小分值范围区间
    if (!zslIsInRange(zsl,range)) return NULL;
    x = zsl->header;
    // 从 header表头中 遍历每个 level 层，从最高层开始
    for (i = zsl->level-1; i >= 0; i--) {
        /* Go forward while *OUT* of range. */
        // 根据 forward 属性往后遍历 每个 score 是否符合条件，找到最近的前一个节点
        while (x->level[i].forward &&
            !zslValueGteMin(x->level[i].forward->score,range))
                x = x->level[i].forward;
    }
    /* This is an inner range, so the next node cannot be NULL. */
    // 获取目标节点
    x = x->level[0].forward;
    serverAssert(x != NULL);
    /* Check if score <= max. */
    // 为了谨慎起见，再判断一下目标节点值是否会大于区间最大值
    if (!zslValueLteMax(x->score,range)) return NULL;
    return x;
}
zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;
    /* If everything is out of range, return early. */
    // 判断给定的区间是否符合 跳跃表 最大最小分值范围区间
    if (!zslIsInRange(zsl,range)) return NULL;
    x = zsl->header;
    // 从 header表头中 遍历每个 level 层，从最高层开始
    for (i = zsl->level-1; i >= 0; i--) {
        /* Go forward while *IN* range. */
        // 根据 forward 属性往后遍历 每个 score 是否符合条件，找到最近的前一个节点
        while (x->level[i].forward &&
            zslValueLteMax(x->level[i].forward->score,range))
                x = x->level[i].forward;
    }
    /* This is an inner range, so this node cannot be NULL. */
    serverAssert(x != NULL);
    /* Check if score >= min. */
    // 为了谨慎起见，再判断一下目标节点值是否会小于区间最小值
    if (!zslValueGteMin(x->score,range)) return NULL;
    return x;
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-skiplist.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！