关于java:Redis源码剖析之跳表skiplist

最近要换新工作了，借着新老工作交替的这段窗口放松了下，所以专栏拖更了，不过我心里毫无愧疚，毕竟没人催更。不过话说回来天天追剧 刷综艺的日子也很是干燥，艳羡你们这些失常下班的人，每天都有正经工作内容，感觉你们过的很空虚。[狗头]

计算机领域有很多种数据结构，数据结构的存在要么是为了节省时间、要么是为了节俭空间，或者二者兼具，所以就有局部数据结构有 工夫换空间，空间换工夫之说 。其实 还有某些以就义准确性来达到节省时间空间的数据结构，像我之间讲过的 bloomfilter 就是其中的典型。而明天要讲的 skiplist 也是一种概率性数据结构，它以一种随机概率降数据组织成多级构造，不便疾速查找。

跳表

到底何为跳表？咱们先来思考下这个场景，假如你有个有序链表，你想看某个特定的值是否呈现在这个链表中，那你是不是只能遍历一次链表能力晓得，工夫复杂度为 O(n)。

可能有人会问为什么不间接用间断存储，咱们还能用二分查找，用链表是想持续保留它批改工夫复杂度低的劣势。那咱们如何优化单次查找的速度？其实思路很像是二分查找，但单链表无奈随机拜访的个性限度了咱们，但二分逐步放大范畴的思路启发了咱们，能不能想什么方法逐步放大范畴？

我是不是能够在原链表之上新建一个链表，新链表是原链表每隔一个节点取一个。假如原链表为 L0，新链表为 L1，L1 中的元素是 L0 中的第 1、3、5、7、9……个节点，而后再建设 L1 和 L0 中各个节点的指针。这样 L1 就能够将 L0 中的范畴放大一半，同理对 L1 再建设新链表 L2……，更高 level 的链表划分更大的区间，确定值域的大区间后，逐级向下放大范畴，如下图。

假如咱们想找 13，咱们能够在 L3 中确定 2 -14 的范畴，在 L2 中确定 8 -14 的范畴，在 L1 中确定 10-14 的范畴，在 L0 中找到 13，整体寻找门路如下图红色门路，是不是比间接在 L0 中找 13 的绿色门路所通过的节点数少一些。

其实这种实现很像二分查找，只不过当时将二分查找的两头点存储下来了，用额定的空间换取了工夫，很容易想到其工夫复杂度和二分查找统一，都是 O(logn)。
小伙子很牛 X 吗，创造了这么牛逼的数据结构，能把有序链表的查找时间复杂度从 O(n)升高到 O(logn)，然而我有个问题，如果链表中插入或者删除了某个节点怎么办？，是不是每次数据变动都要重建整个数据结构？

其实不用，咱们不须要严格保障两两层级之间的二分之一的关系，只须要概率上为二分之一 就行，删除一个节点好说，间接把某个层级中对应的改节点删掉，插入节点时，新节点以指数递加的概率往下层链表插入即可。比方 L0 中 100% 插入，L1 中以 1 / 2 的概率插入，如果 L1 中插入了，L2 中又以 1 / 2 的概率插入…… 留神，只有高 Level 中有的节点，低 Level 中肯定有，但高 Level 链表中呈现的概率会随着 level 指数递加 ，最终跳表可能会长这个样子。

咱们就这样从新创造了 skiplist。

Redis 中的跳表

Redis 为了提供了有序汇合 (sorted set) 相干的操作(比方 zadd、zrange)，其底层实现就是 skiplist。咱们接下来看下 redis 是如何实现 skiplist 的。

typedef struct zskiplist {
    struct zskiplistNode *header, *tail; // 头尾指针 
    unsigned long length;   // skiplist 的长度  
    int level;  // 最高多少级链表 
} zskiplist;

咱们先来看下 redis 中 zskiplist 的定义，没啥内容，就头尾指针、长度和级数，重点还是在 zskiplistNode 中。zskiplistNode 中是有前向指针的，所以 Level[0]其实是个双向链表。

typedef struct zskiplistNode {
    sds ele;   // 节点存储的具体值 
    double score;   // 节点对应的分值 
    struct zskiplistNode *backward; // 前向指针
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 每一层的后向指针 
        unsigned long span;  // 到下一个节点的跨度 
    } level[];} zskiplistNode;

redis 中的 skiplist 实现略微和咱们上文中讲的不大一样，它并不是简略的多级链表的模式，而是间接在 zskiplistNode 中的 level[]将不同 level 的节点的关联关系组织起来，zskiplist 的构造可视化如下。

跳表的操作

晓得了 zskiplist 的结构，咱们来看下其几个次要操作。

新建跳表

/* 创立跳表 */
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;

    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;
    zsl->length = 0;
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);  // 创立头节点
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

创立跳表就比较简单了，间接创立一个空的节点做为头节点。

/* 在跳表中插入一个新的节点, */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    x = zsl->header;
    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];
        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;
    }
    /* skiplist 中不会呈现反复的元素，但咱们容许反复的分值，因为如果是调用 zslInsert()的话，不会呈现反复插入两
     * 个雷同的元素，因为在 zslInsert()中曾经判断了 hash 表中是否存在 */
    level = zslRandomLevel();  // 生成一个随机值，确定最高须要插入到第几级链表里  
    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);  // 为插入的数据创立新节点  
    for (i = 0; i < level; i++) {x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        /* 插入新节点后须要更新前后节点对应的 span 值 */
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    /* 为其余 level 减少 span 值，因为在原有俩节点之间插入了一个新节点 */
    for (i = level; i < zsl->level; i++) {update[i]->level[i].span++;
    }

    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    zsl->length++;
    return x;
}
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) // ZSKIPLIST_P == 0.25
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

数据插入就略微简单些，须要新建节点，而后确定须要在哪些 level 中插入新节点，还要更新前节点中各个 level 的 span 值。这里额定留神下 zslRandomLevel，zslRandomLevel 是以 25% 的概率决定是否将单个节点搁置到下一层，而不是 50%。

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {if (update[i]->level[i].forward == x) {update[i]->level[i].span += x->level[i].span - 1;  // 删除节点须要批改 span 的值
            update[i]->level[i].forward = x->level[i].forward;
        } else {update[i]->level[i].span -= 1;
        }
    }
    if (x->level[0].forward) {x->level[0].forward->backward = x->backward;
    } else {zsl->tail = x->backward;}
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}

/* 从 skiplist 中删除 ele，如果删除胜利返回 1，否则返回 0. 
 *
 * 如果节点是 null，须要调用 zslFreeNode()开释掉该节点，否则只是把指向 sds 的指针置空，这样
 * 后续其余的节点还能够持续应用这个 sds*/
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {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;
    }
    /* 可能有多个节点有雷同的 socre，都必须找进去并删除  */
    x = x->level[0].forward;
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) {zslDeleteNode(zsl, x, update);
        if (!node)
            zslFreeNode(x);
        else
            *node = x;
        return 1;
    }
    return 0; /* not found */
}

数据的删除也很简略，很相似于单链表的删除，但同时须要更新各个 level 上的数据。

其余代码就比拟多，晓得了 skiplist 的具体实现，其余相干操作的代码也就比拟容易想到了，我这里就不在列举了，有趣味能够查阅下 t_zset.c

Redis 为什么应用 skiplist 而不是均衡树

Redis 中的 skiplist 次要是为了实现 sorted set 相干的性能，红黑树当然也能实现其性能，为什么 redis 作者当初在实现的时候用了 skiplist 而不是红黑树、b 树之类的均衡树？而且显然红黑树比 skiplist 更节俭内存啊！Redis 的作者 antirez 也已经亲自回应过这个问题，原文见 https://news.ycombinator.com/item?id=1171423

我大抵翻译下：

1. skiplist 并不是特地耗内存，只须要调整下节点到更高 level 的概率，就能够做到比 B 树更少的内存耗费。
2. sorted set 可能会面对大量的 zrange 和 zreverange 操作，跳表作为单链表遍历的实现性能不亚于其余的均衡树。
3. 实现和调试起来比较简单。例如，实现 O(log(N))工夫复杂度的 ZRANK 只须要简略批改下代码即可。

本文是 Redis 源码分析系列博文，同时也有与之对应的 Redis 中文正文版，有想深刻学习 Redis 的同学，欢送 star 和关注。
Redis 中文注解版仓库：https://github.com/xindoo/Redis
Redis 源码分析专栏：https://zxs.io/s/1h
如果感觉本文对你有用，欢送 一键三连。