skiplist 记录

一次典型的跳表查询过程

skiplist上的查找路径展示

　　 skiplist正是受这种多层链表的想法的启发而设计出来的。实际上，按照上面生成链表的方式，上面每一层链表的节点个数，是下面一层的节点个数的1/P(redis中P为0.25)，这样查找过程就非常类似于一个二分查找，使得查找的时间复杂度可以降低到O(log n)

核心数据结构

/*
 * 跳跃表节点
 在同一个跳跃表中， 各个节点保存的成员对象必须是唯一的， 但是多个节点保存的分值却可以是相同的： 分值相同的节点将按照成员
 对象在字典序中的大小来进行排序， 成员对象较小的节点会排在前面（靠近表头的方向）， 而成员对象较大的节点则会排在后面（靠
 近表尾的方向）。
 */
typedef struct zskiplistNode {

     // 表头节点也有后退指针、分值和成员对象， 不过表头节点的这些属性都不会被用到
    // member 对象      节点所保存的成员对象。
    robj *obj;
    // 分值 // 分值 在跳跃表中，节点按各自所保存的分值从小到大排列。 跳跃表中的所有zskiplistNode节点都按分值从小到大来排序。
    double score;
    // 后退指针
    struct zskiplistNode *backward;
    /* 层
     节点中用 L1 、 L2 、 L3 等字样标记节点的各个层， L1 代表第一层， L2 代表第二层，以此类推。每个层都带有两个属性：前进指针和跨度。
     跳跃表节点的 level 数组可以包含多个元素， 每个元素都包含一个指向其他节点的指针， 程序可以通过这些层来加快访问其他节点的
     速度， 一般来说， 层的数量越多， 访问其他节点的速度就越快。  
     每次创建一个新跳跃表节点的时候， 程序都根据幂次定律 （power law，越大的数出现的概率越小） 随机生成一个介于 1 和 32 之间
     的值作为 level 数组的大小， 这个大小就是层的“高度”。
     */
    struct zskiplistLevel {
        // 前进指针
        // 前进指针 前进指针用于访问位于表尾方向的其他节点  当程序从表头向表尾进行遍历时，访问会沿着层的前进指针进行。
        struct zskiplistNode *forward;
        // 跨度
        // 这个层跨越的节点数量  跨度则记录了前进指针所指向节点和当前节点的距离。
        unsigned int span;
    } level[]; //创建新的zskiplistNode节点的时候，level层数数组[]大小时随机产生的，见zslInsert->zslRandomLevel
} zskiplistNode; //存储在zskiplist跳跃表结构中

/*
 * 跳跃表
 */
/*
header ：指向跳跃表的表头节点。
?tail ：指向跳跃表的表尾节点。
?level ：记录目前跳跃表内，层数最大的那个节点的层数（表头节点的层数不计算在内）。
?length ：记录跳跃表的长度，也即是，跳跃表目前包含节点的数量（表头节点不计算在内）。
*/ 
typedef struct zskiplist { //zslCreate
    // 表头节点和表尾节点 // 头节点，尾节点  注意在创建zskiplist的时候默认有创建一个头节点，见zslCreate
    struct zskiplistNode *header, *tail;
    // 表中节点的数量
    unsigned long length;
    // 目前表内节点的最大层数 level 属性则用于在 O(1) 复杂度内获取跳跃表中层高最大的那个节点的层数量， 注意表头节点的层高并不计算在内。
    //创建zskiplist的时候zslCreate中默认置1
    int level; //创建新的zskiplistNode节点的时候，level层数数组[]大小时随机产生的，见zslInsert->zslRandomLevel
} zskiplist;

 

插入实现：

/*
 * 创建一个成员为 obj ，分值为 score 的新节点，
 * 并将这个新节点插入到跳跃表 zsl 中。
 * 函数的返回值为新节点。
 * T_wrost = O(N^2), T_avg = O(N log N)
 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
     //定义最大的（保证不会越界）更新节点前一个节点数组及相关节点
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
     //访问过程中经过的所有层的span相加即为rank的值
    unsigned int rank[ZSKIPLIST_MAXLEVEL]; 
    int i, level;
    redisAssert(!isnan(score));
    // 在各个层查找节点的插入位置
    // T_wrost = O(N^2), T_avg = O(N log N)
    x = zsl->header;
     //从最简入手，最高层节点最少
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        // 如果 i 不是 zsl->level-1 层
        // 那么 i 层的起始 rank 值为 i+1 层的 rank 值
        // 各个层的 rank 值一层层累积
        // 最终 rank[0] 的值加一就是新节点的前置节点的排位
        // rank[0] 会在后面成为计算 span 值和 rank 值的基础      
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 沿着前进指针遍历跳跃表
        // T_wrost = O(N^2), T_avg = O(N log N)
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                // 比对分值
                (x->level[i].forward->score == score &&
                // 比对成员， T = O(N)
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            // 记录沿途跨越了多少个节点
            rank[i] += x->level[i].span;
            // 移动至下一指针
            x = x->level[i].forward;
        }
        // 记录将要和新节点相连接的节点
        update[i] = x;
    }
    /* we assume the key is not already inside, since we allow duplicated
     * scores, and the re-insertion of score and redis object should never
     * happen since the caller of zslInsert() should test in the hash table
     * if the element is already inside or not. 
     * zslInsert() 的调用者会确保同分值且同成员的元素不会出现，
     * 所以这里不需要进一步进行检查，可以直接创建新元素。
     */
    // 获取一个随机值作为新节点的层数
    // T = O(N)
    level = zslRandomLevel();//随机生成高度
    // 如果新节点的层数比表中其他节点的层数都要大
    // 那么初始化表头节点中未使用的层，并将它们记录到 update 数组中
    // 将来也指向新节点
    if (level > zsl->level) {
        // 初始化未使用层
        // T = O(1) //大于LEVEL设置RANK为0，为什么？因为最高前后就它一根独苗
        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,obj);
    // 将前面记录的指针指向新节点，并做相应的设置
    // T = O(1) //如链表一样，修改前向节点
    for (i = 0; i < level; i++) {        
        // 设置新节点的 forward 指针
        x->level[i].forward = update[i]->level[i].forward;       
        // 将沿途记录的各个节点的 forward 指针指向新节点                 /    /插入新节点
        update[i]->level[i].forward = x;
        /* update span covered by update[i] as x is inserted here */
        // 计算新节点跨越的节点数量
         //从第一个节点到插入节点后一个节点的距离（AC距离）是update[i]->level[i].span 
         //说明： A(First)   B(insert pos)   C（forward）
        //插入节点到后面的节点的距离（B和A的距离）是rank[0] - rank[i]
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 更新新节点插入之后，沿途节点的 span 值
        // 其中的 +1 计算的是新节点
        //新增加的节点后面的距离一定是(rank[0] - rank[i]) + 1
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    /* increment span for untouched levels */
    // 未接触的节点的 span 值也需要增一，这些节点直接从表头指向新节点
    // T = O(1)//处理高层的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;
}

/* Returns a random level for the new skiplist node we are going to create. The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL
 * (both inclusive), with a powerlaw-alike distribution where higher levels are less likely to be returned. 
 *
 * 返回值介乎 1 和 ZSKIPLIST_MAXLEVEL 之间（包含 ZSKIPLIST_MAXLEVEL），
 * 根据随机算法所使用的幂次定律，越大的值生成的几率越小。
 *有 1/2 的概率返回 1、1/4 的概率返回 2、1/8的概率返回 3，以此类推
 * T = O(N)
 */
int zslRandomLevel(void) {
    int level = 1;
    //0和65535之间的数和ZSKIPLIST_P * 0xFFFF=0.25*65535比较，限定概率
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

 

插入包含如下几个操作：1、查找到需要插入的位置   2、申请新的结点    3、调整指针。

 

 

 span 属性，用来 表示从前一个节点沿着当前层的 forward 指针跳到当前这个节点中间会跳过多少个节点

 以节点19插入为例，其中
黑色箭头的表示的跨度为update[i]->level[i].span
蓝色箭头表示的跨度为rank[0] - rank[i]即节点19在level_0的update[0]为11，在level_1的update[1]为7，rank[0] - rank[i]为节点7与节点11之间的跨度
绿色箭头表示的跨度为节点19到节点37的span

删除

    删除操作类似于插入操作，包含如下3步：1、查找到需要删除的结点 2、删除结点  3、调整指针

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank 
 * 
 * 内部删除函数，
 * 被 zslDelete 、 zslDeleteRangeByScore 和 zslDeleteByRank 等函数调用。
 *
 * T = O(1)
 */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    // 更新所有和被删除节点 x 有关的节点的指针，解除它们之间的关系
    // T = O(1)
    for (i = 0; i < zsl->level; i++) {
        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;
        }
    }
    // 更新被删除节点 x 的前进和后退指针
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    // 更新跳跃表最大层数（只在被删除节点是跳跃表中最高的节点时才执行）
    // T = O(1)
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;

    // 跳跃表节点计数器减一
    zsl->length--;
}
/* Delete an element with matching score/object from the skiplist. 
 * 从跳跃表 zsl 中删除包含给定节点 score 并且带有指定对象 obj 的节点。
 * T_wrost = O(N^2), T_avg = O(N log N)
 */
int zslDelete(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;
    // 遍历跳跃表，查找目标节点，并记录所有沿途节点
    // T_wrost = O(N^2), T_avg = O(N log N)
    x = zsl->header;
     // 遍历所有层，记录删除节点后需要被修改的节点到 update 数组
    for (i = zsl->level-1; i >= 0; i--) {
        // 遍历跳跃表的复杂度为 T_wrost = O(N), T_avg = O(log N)
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                // 比对分值
                (x->level[i].forward->score == score &&
                // 比对对象，T = O(N)
                compareStringObjects(x->level[i].forward->obj,obj) < 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. 
     * 检查找到的元素 x ，只有在它的分值和对象都相同时，才将它删除。
     */// 因为多个不同的 member 可能有相同的 score 
    // 所以要确保 x 的 member 和 score 都匹配时，才进行删除
    x = x->level[0].forward;
    if (x && score == x->score && equalStringObjects(x->obj,obj)) {
        // T = O(1)
        zslDeleteNode(zsl, x, update);
        // T = O(1)
        zslFreeNode(x);
        return 1;
    } else {
        return 0; /* not found */
    }

    return 0; /* not found */
}

假设SkipList中有n个元素，第L层级元素个数的期望是1/p个；每个元素出现在L层的概率是p^(L-1), 那么第L层级元素个数的期望是 n * (p^L-1)；得到1 / p =n * (p^L-1)

1 / p = n * (p^L-1)
n = (1/p)^L
L = log(1/p)^n
所以我们应该选择MaxLevel = log(1/p)^n
定义：MaxLevel = L(n) = log(1/p)^n

Q：redis中为啥不用红黑树二用跳表

1 内存占用方面跳表比红黑树多，但是多的内存很有限
2 实现比红黑树简单
3 跟红黑树更方便的支持范围查询

Q：跳表层数上限为啥是32？

根据前面的随机算法当level[0]有2的64次方个节点时，才能达到32层，因此层数上限是32完全够用了。