1.skiplist介绍

• 定义：跳跃表是一个有序链表，其中每个节点包含不定数量的链接，节点中的第i个链接构成的单向链表跳过含有少于i个链接的节点。
• 跳跃表支持平均O(logN)，最坏O(N)复杂度的节点查找，大部分情况下，跳跃表的效率可以和平衡树相媲美。
• 跳跃表在redis中当数据较多时作为有序集合键的实现方式之一。

接下来，还是举个有序集合键的例子：有序集合键详解

127.0.0.1:6379> ZADD score 95.5 Mike 98 Li 96 Wang //socre是一个有序集合键

(integer) 3

127.0.0.1:6379> ZRANGE score 0 -1 WITHSCORES//所有分数按从小到大排列，每一个成员都保存了一个分数

1) "Mike"

2) "94.5"

3) "Wang"

4) "96"

5) "Li"

6) "98"

127.0.0.1:6379> ZSCORE score Mike //查询Mike的分值

"94.5"

2.结构体分析

跳表节点

typedef struct zskiplistNode {

sds ele; //保存成员数据的字符串

double score; //分值

struct zskiplistNode *backward;//后退指针

struct zskiplistLevel {//索引层结构

struct zskiplistNode *forward;//前进指针

unsigned long span;//跨度

} level[];

} zskiplistNode;

一个zskiplistNode 包含多个zskiplistLevel 组成 level[]，每一个zskiplistLevel 都是其对应层的一个索引节点

typedef struct zskiplist {

struct zskiplistNode *header, *tail;//头节点 尾部节点

unsigned long length;//节点数

int level;//最大层级

} zskiplist;

3.主要函数分析

3.1 创建跳跃表

zskiplist *zslCreate(void) {

int j;

zskiplist *zsl;

zsl = zmalloc(sizeof(*zsl));

zsl->level = 1;

zsl->length = 0;

zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);//创建一个层级是32，分数是0，成员数据是NULL的头节点

for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {// 初始化 头节点的层级数组的元素，前进元素指向NULL，跨度为0

zsl->header->level[j].forward = NULL;

zsl->header->level[j].span = 0;

}

zsl->header->backward = NULL;//头节点的后退指针指向NULL

zsl->tail = NULL;//尾节点指向 NULL

return zsl;

}

创建一个层级是32，分数是0，成员数据是NULL的头节点 是要注意的地方

3.2 插入节点

跳跃表的中每一个层级 相当于一层索引，称为索引层，插入新节点时 要同时插入 相关的索引层，将插入索引层的节点叫做索引节点。

一个跳表节点包含一个level[]，即多个索引节点

在 zsl 跳跃表中插入 分数为score，成员数据为ele的节点

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--) {//遍历所有层级 收集 update[] 和rank[]数据 步骤1

/* store rank that is crossed to reach the insert position */

//rank[i] 代表序号i的i+1层的跨越节点总数，下面的逻辑是因为 每次 循环 都是从 i+1层的x节点开始， 相当于rank[i]的初始值是rank[i+1]

//zsl->level-1==0 说明 是从 头节点开始计算，于rank[i]的初始值是0

rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];

while (x->level[i].forward &&

(x->level[i].forward->score < score |||//只要分数小于 参数score 向后遍历

(x->level[i].forward->score == score &&//分数相等 且参数key字符串值相大于等于下个节点

sdscmp(x->level[i].forward->ele,ele) < 0)))

{//重点：这里说明分数相同时，按照key的字符串值 从小到大排列

rank[i] += x->level[i].span;

x = x->level[i].forward;

}

update[i] = x;//保存了 每一个层分值小于等新节点score的最后一个节点，相当于保存了 每个索引层中 新节点的前置节点

//update[0] 就是要插入跳表节点 的前置跳表节点

}

//update[0] 代表 level1 层级中的跳表节点，目前的这个跳跃表结构 level1是最小的索引层，一定包含了所有数据节点的索引节点，

//所以update[0] 就是要插入节点 的前置节点。

// 执行步骤1 是因为 插入新的节点 可能需要在多个层插入，收集 update[]方便插入，收集rank[]数据 ，方便计算 索引节点的跨度

/* 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. */

//注释：允许相同的分数，不允许有相同的成员数据， 调用者在执行前应该检查 利用hash table结构

//实际上 调用 zslInsert()前已经做了这样的检查，可以放心执行。

level = zslRandomLevel();//获取 level 返回1的概率50%，2：25% 3:12.5% 以此类推 数越大概率越小，最大返回32

if (level > zsl->level) {// 如果 获取的索引级别当前的值大

for (i = zsl->level; i < level; i++) {//初始化新增的索引级别

rank[i] = 0;//跨度都是0

update[i] = zsl->header;//这个层级相对于 新增节点的前置节点都是header,因为这个层级还没加入其它索引节点

update[i]->level[i].span = zsl->length;//索引节点的跨度是总节点个数

}

zsl->level = level;

}

x = zslCreateNode(level,score,ele);

for (i = 0; i < level; i++) {//遍历update[]插入新节点

//插入索引节点

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]);//(rank[0] - rank[i]) 计算的是 x->level[i]与前置节点跨度差，

//update[i]->level[i].span 是 x->level[i] 到x->level[i]->forward 节点的跨度，就像 我剪短一根绳子 后一段长度= 总长度 -前1一段长度

update[i]->level[i].span = (rank[0] - rank[i]) + 1;//增加一个节点 自然要加1

}

/* increment span for untouched levels */

for (i = level; i < zsl->level; i++) {//比较高的层级没有插入新的索引节点，但实际上总体增加了一个节点，对应层级的前置节点跨度需要加1

update[i]->level[i].span++;

}

//前面分析过 update[0]是前置跳表节点

x->backward = (update[0] == zsl->header) ? NULL : update[0];

if (x->level[0].forward)//设置x的下个节点的后退节点是x

x->level[0].forward->backward = x;

else//x没有下个节点 设置 zsl的尾部节点是x

zsl->tail = x;

zsl->length++;

return x;

}

3.2 删除节点

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {

int i; //在执行删除之前 获取 update，这里的update代表的是指向zskiplistNode 数组的指针，与插入中update[]相同

//都保存了 x 在每个索引层级的前置跳表节点

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) {//删除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--;

//删除节点后没有 回收内存，推断在一个完整的删除中，有回收的处理。

}

3.3 获取节点排名

根据 节点key和分数获取节点排名， 要求key对应的节点必须存在，这个节点的分数 必须大于等于传入的分数

unsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {

zskiplistNode *x;

unsigned long rank = 0;

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))) {//注意这里 多个了 ‘=’ 也就是说遇到 key相等的情况还向后遍历。

//与插入不同 插入 是为了找前置节点 查找是 是为了获取key对应的节点，所以这里要多向后遍历一步

rank += x->level[i].span;

x = x->level[i].forward;

}

/* x might be equal to zsl->header, so test if obj is non-NULL */

if (x->ele && sdscmp(x->ele,ele) == 0) {//key必须相等，就是说key对应的节点必须存在。

return rank;

}

}

return 0;

}

逻辑比较简单 每个节点都记录了跨度，累加遍历节点跨度就是排名，执行 遍历跳表 找到key值相等 分数大于等于传入分数的节点，累加排名，返回排名。

3.4范围查找

查找指定范围内的第一个节点

zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec *range) {//range 表示一个返回 包含 一个最小值 和一个最大值

zskiplistNode *x;

int i;

/* If everything is out of range, return early. */

if (!zslIsInRange(zsl,range)) return NULL;//检查 zsl中的节点是否在 range表示的范围内

x = zsl->header;

for (i = zsl->level-1; i >= 0; i--) {

/* Go forward while *OUT* of range. */

while (x->level[i].forward &&//下个节点有效

!zslValueGteMin(x->level[i].forward->score,range))// 下个节点 小于 range中的最小值

x = x->level[i].forward;

}

// x的下个节点刚好大于range中的最小值,前进一个节点

/* 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; //检查是否 大于 range中的最大值

return x;

}

zslLastInRange ：查找指定范围内的最后一个节点， 某些判断与zslFirstInRange相反 ， 处理逻辑相同。就不详细分析了