【redis源码学习】跳跃表

2021-12-23 15:16:42 浏览数 (1)

之前手写过跳表，所以概念性的东西就不多提，直接上链接：数据结构（9）-- 跳表

在redis中跳表主要应用于有序集合的底层实现。 这个别人怎么讲都意义不大，自己动手去写一下才知道其中的妙处与不容易。没有看起来那么好写。

然后，来学习一下redis中是如何实现跳表的，看看自己跟大师的天差地别。

文章目录
  • 跳表整体概览
  • 跳跃表节点
  • 跳跃表结构
  • 创建跳跃表
    • 随机数获取
    • 创建跳跃表结构
    • 创建跳跃表节点
    • 插入节点
    • 删除节点
    • 删除整表

跳表整体概览

1、由多层构成。 2、有一个头节点，指向一个64层结构。还有个尾结点。 3、除头结点外，层次最高的层数记录在 level 中。 4、length记录了最底层链表的长度。 5、红色部分有往回指的。下面来看这个红色部分。

跳跃表节点

代码语言：javascript复制
/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
    sds ele;	//每个节点的 ele 存储有序集合的成员member值
    double score;	//score存储成员分值。所有节点的分值都是从小到大排序的，遇到分值相同的则按ele的字典序排序。
    struct zskiplistNode *backward;	//后退指针，上面红色部分最顶层
		//下面是粉红色部分
    struct zskiplistLevel {
        struct zskiplistNode *forward;	//指向本层后一个节点
        unsigned long span;	//两个节点之间跳过了多少个节点
    } level[];	//那这个就是存一层一层的节点咯。生成每一层的管理员节点（意会一下）时，随机生成一个1~64的值。值越大出现的概率越低。
} zskiplistNode;

跳跃表结构

链表都是有结构 节点 组成的，跳跃表出自链表，自然也有结构。

代码语言：javascript复制
//图中蓝色部分
typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

局势已经有点明朗了吧。

创建跳跃表

随机数获取

略微抽象哈。

代码语言：javascript复制
/* 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. */
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))	//ZSKIPLIST_P  = 0.25，0xFFFF = 1111 1111 1111 1111
        level  = 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;	//ZSKIPLIST_MAXLEVEL = 64
}

所以这个循环的意思就是，取一个随机数random，取这个值的低16位（共32位）作为x，当x小于0.25倍的0xFFFF（十进制为65535）时，level 加一。

如果我没猜错，这个每次成功加一的概率为 0.25，可以口算，但是笔算我就不会写了。

节点 n 层高的概率为 (1-p)*p^(n-1)，所以节点的期望层高为 1/(1-p) = 1.33。

好低啊。。。

创建跳跃表结构

就是那个蓝色部分：

代码语言：javascript复制
/* Create a new skiplist. */
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;
}

不难理解吧，跟链表创建头结点一样的。

创建跳跃表节点

初始化操作总是那么的平平无奇哈。后面的增删改查才是重头戏！！！

代码语言：javascript复制
/* Create a skiplist node with the specified number of levels.
 * The SDS string 'ele' is referenced by the node after the call. */
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
    zskiplistNode *zn =
        zmalloc(sizeof(*zn) level*sizeof(struct zskiplistLevel));
    zn->score = score;
    zn->ele = ele;
    return zn;
}

插入节点

插入节点的步骤如下： 1、寻找插入位置 2、调增跳跃表高度 3、插入节点 4、调整backward

看似不难，谁写谁知道。

代码语言：javascript复制
/* Insert a new node in the skiplist. Assumes the element does not already
 * exist (up to the caller to enforce that). The skiplist takes ownership
 * of the passed SDS string 'ele'. */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    //update：保存被插入节点每层的前一个节点，直接看很难明白为什么要这个东西，自己写过一遍就知道这是个好东西了
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    //rank：记录当前层从header到update[i]所需步长，更新span的时候需要用到
    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;
    }
    /* 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. */
    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;

        /* update span covered by update[i] as x is inserted here */
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i])   1;
    }

    /* increment span for untouched levels */
    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;
}

我们用这个示例来看这个跳跃表的插入循环：

找这么张图不容易哈。 我们现在往里面插入 score=27，level=3 的数据。

代码语言：javascript复制
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;
}

第一次进入for循环： 1）x为头，i=1； 2）rank[1] = 0； 3）进入while循环； 4）rank[1] = 1；x成为第一个节点（也就是update(1)那边） 5）由于第一个节点的L1层后继无人，所以被迫终止循环。update[i]=x；

第二次进入for循环（层级下降）： 1）x为第一个节点，i=0； 2）rank[0]=rank[1]=1； 3）进入循环，这时已经下降一层了。 4）rank[0] = 2； 5）x继续往后，进入到第二个节点（也就是update[0]那边了） 6）由于插入score干不过后面那个，所以被迫退出循环，update[0]=x；

第三次进入for循环？nonono，莫得了、

接下来干嘛？调整最大高度。

代码语言：javascript复制
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;
}

rank用来记录头节点到update的span，不要忘了。

由于目标level=2，所以会进入分支。 1）rank[2]=0，因为update[2]是头结点。 2）update[2]是头结点。 3）先给level赋个值，后面会调整。

接下来干嘛？前戏都做完了，接下来插入节点。

代码语言：javascript复制
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;

        /* update span covered by update[i] as x is inserted here */
    x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
    update[i]->level[i].span = (rank[0] - rank[i])   1;
}

如果新插入的列太矮，那比较高的那一部分就不用去转接，只需要span 1即可。

代码语言：javascript复制
/* increment span for untouched levels */
for (i = level; i < zsl->level; i  ) {
    update[i]->level[i].span  ;
}

还剩最后一段，调整backword。

代码语言：javascript复制
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;

真是庖丁解牛。我自己写的时候没有想这么清楚，就搞得有点乱。

查改就不说了吧，上面有查了，改就更不多说了吧，再看一下删。

删除节点

代码语言：javascript复制
/* Internal function used by zslDelete, zslDeleteRangeByScore and
 * zslDeleteRangeByRank. */
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;
            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--;
}

看完增，删就是一个缩减版的逆操作，应该不难理解吧。

删除整表

这里的英文解释挺详尽的了，代码也很清晰。从第0层开始，通过forward向后遍历，一个一个回收内存。节点都回收完了，再回收表结构。

代码语言：javascript复制
/* Delete an element with matching score/element from the skiplist.
 * The function returns 1 if the node was found and deleted, otherwise
 * 0 is returned.
 *
 * If 'node' is NULL the deleted node is freed by zslFreeNode(), otherwise
 * it is not freed (but just unlinked) and *node is set to the node pointer,
 * so that it is possible for the caller to reuse the node (including the
 * referenced SDS string at node->ele). */
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;
    }
    /* 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 = 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 */
}

今天先到这里吧。大家都是按部就班的，字符串，压缩表，哈希表。。。。我反而觉得压缩表不如跳跃表来的有意思哈哈。

编程算法 云数据库Redis®

0 人点赞