redis源码分析之六基础的数据结构skiplist

一、skiplist 跳表

跳表这个数据结构是新生的，在学习数据结构的时候儿是没有这个的。当然，也可以理解成是对数据结构的进一步的封装，这样理解的话，可能就会更准确一些。为什么叫跳表？想想生活中跳的动作，一般人走路是一步一步的走，而如果跳跃的话，一下子可以走好几步，但是付出的代价就是要多费些力气。 其实跳表也是如此，正常的链表list，访问的时候儿是从头到尾（或者反过来）一条条的遍历，而跳表由于多了一些特殊的指针，可以一下子就访问距离自己更远的节点，并通过一些状态，来判定是否满足要求。如果满足就不再向前跳。举一个最基本的例子，比如普通的链表中增加一个指针，这个指针始终指向链表的中间节点。假如这些链表节点通过某种状态进行排序，那么，是不是就回到了查找算法的二分法，如果进一步在增加几个指针，都指向余下的部分的中间节点，这是不是就是二分查找。 但付出的代价就是，节点增加了不少指针，增大了空间的使用，另外就是导致整个链表的操作变得复杂了。看下图的图：

普通链表：

跳表：

实际的跳表也是这个思想，增加指针，按某种特征或者状态排序，设置步长，不是简单的二分。那么这个跳表有什么实际意义么？这得看你怎么看。在基础的数据结构中，处理大数据的增删改查，基本都是用树，各种树就应运而生，但是有一个实际情况，对于绝大多数的程序员来说，面试时能手撸红黑树,AVL树代码的，估计很少，而这很少的人中，绝大多数都是提前复习过，并不是说在日常工作中常用，拿起来就写。这也就是网上的人们经常说的“面试造火箭，工作拧螺丝”的原因。 而跳表就简单了，只要掌握了原理，看一看别人的实现，基本就没有什么大问题了。不过有一点要注意，跳表是一个“概率”型的数据结构，它的插入是通过概率来计算的，通过随机产生一个1~MAX_LEVEL之间的高度，然后将节点插入，所有链中相同的元素可以称为“同位素”。在Redis中定义如下：

#define ZSKIPLIST_MAXLEVEL 64 /* Should be enough for 2^64 elements */ #define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */

跳表的空间复杂度为 O(n)，跳表的查询复杂度为，平均查找O(logN)、最坏O(N)。跳表主要应用在ZSet和集群内部数据结构。 说明：跳表由William Pugh 1989年发明。论文《Skip lists: a probabilistic alternative to balanced trees》可以去网上查找。有兴趣建议看原版论文，这样会更清楚设计的原理和方向。

二、源码分析

说一千道一万，还得看源码(3.0在redis.h中，4.0在server.h中)：

1、定义

typedef struct zskiplistNode { sds ele;//Zset成员对象，3.0是robj double score;//分值 struct zskiplistNode * backward; //后退指针，最下层 struct zskiplistLevel { struct zskiplistNode * forward;//前进指针，同一层 unsigned long span;//跨度 } level[]; } zskiplistNode; typedef struct zskiplist { struct zskiplistNode * header, * tail;//头尾节点指针 unsigned long length;//节点数量 int level;//最大节点层数 } zskiplist; typedef struct zset { dict * dict;//保存值和Score zskiplist * zsl;//跳表指针 } zset;

跳表的功能也不外乎常见的几个功能，插入、删除和查找。使用 sds ele的目的是和其它语言中的功能保持一致，也就是保持KEY的单一性，不允许重复插入。分值Score是通过命令传入，保存到字典中，使用时直接到zsetScore函数中获取即可。

2、插入

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 *zslCreateNode(int level, double score, sds ele) { zskiplistNode * zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel)); zn->score = score; zn->ele = ele; return zn; } zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) { //定义最大的（保证不会越界）更新节点前一个节点数组及相关节点 zskiplistNode * update[ZSKIPLIST_MAXLEVEL], * x; //指定（最大的随机层级）排名数组，最大兼容保证不会越界 //访问过程中经过的所有层的span相加即为rank的值 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的位置，即插入最近的排名（小于） //初始化，如果为最高层，则为0，否则为上一层的rank的值 //这是后面用来计算span和rank的基础 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))) { //持续计算span排名 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(); //大于LEVEL设置RANK为0，为什么？因为最高前后就它一根独苗，看一下图就明白了。 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 */ //从第一个节点到插入节点后一个节点的距离（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]); //新增加的节点后面的距离一定是(rank[0] - rank[i]) + 1 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++; } //处理从后向前遍历的指针，第一个时直接指向头 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; //0和65535之间的数和ZSKIPLIST_P * 0xFFFF=0.25*65535比较，限定概率 while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) level += 1;//如果小于成功，则继续 return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL; }

创建跳表没有什么，就是分配内存同时把命令传进来的相关参数进行存储。插入时分为两种情况，即在Level范围内和大于Level，大于时要注意处理Span和Update[i].随机函数其实就对随机范围通过P值来调整。确定是返回0或者1，来间接决定Level的高度。 另外还有一个重载的zslInsert，是用在ziplist编码时使用的。

3、删除

在zremrangeGenericCommand函数中会分别对不同的删除进行处理，看skiplist编码部分，有三种删除方式：

void zremrangeGenericCommand(client *c, int rangetype) { ...... else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) { zset * zs = zobj->ptr; switch(rangetype) { case ZRANGE_RANK: deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict); break; case ZRANGE_SCORE: deleted = zslDeleteRangeByScore(zs->zsl,&range,zs->dict); break; case ZRANGE_LEX: deleted = zslDeleteRangeByLex(zs->zsl,&lexrange,zs->dict); break; } if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (dictSize(zs->dict) == 0) { dbDelete(c->db,key); keyremoved = 1; } } ...... } unsigned long zslDeleteRangeByRank(zskiplist * zsl, unsigned int start, unsigned int end, dict * dict) { zskiplistNode * update[ZSKIPLIST_MAXLEVEL], * x; unsigned long traversed = 0, removed = 0; int i; x = zsl->header; //查找要删除的节点update节点 for (i = zsl->level-1; i >= 0; i--) { while (x->level[i].forward && (traversed + x->level[i].span) < start) { traversed += x->level[i].span; x = x->level[i].forward; } update[i] = x; } traversed++; x = x->level[0].forward; while (x && traversed <= end) { zskiplistNode * next = x->level[0].forward; zslDeleteNode(zsl,x,update);//删除 dictDelete(dict,x->ele); zslFreeNode(x); removed++; traversed++; x = next; } return removed; } void zslDeleteNode(zskiplist * zsl, zskiplistNode * x, zskiplistNode ** update) { int i; for (i = 0; i < zsl->level; i++) { //update节点的向前节点forward即为删除节点，则处理相关数据 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 { //span-1，因为本层未出现 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--; } //删除单点 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 * / }

删除和插入从原理上来讲是一致的，都是要处理前后的指针数据，不过是一个增加，一个删除。

4、查找 跳跃表提供了根据排名rank查询和间接分数查询的API，前面提到过，跳表其实是分层查找的，如果能在同一层查到，那效率是最高的，最坏的情况下是全部遍历所有的层。

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))) { 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) { return rank; } } return 0; } /* Finds an element by its rank. The rank argument needs to be 1-based. */ zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) { zskiplistNode * x; unsigned long traversed = 0; int i; x = zsl->header; for (i = zsl->level-1; i >= 0; i--) { while (x->level[i].forward && (traversed + x->level[i].span) <= rank) { traversed += x->level[i].span; x = x->level[i].forward; } if (traversed == rank) { return x; } } return NULL; } 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; 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)) 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; } /* Find the last node that is contained in the specified range. * Returns NULL when no element is contained in the range. */ 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; for (i = zsl->level-1; i >= 0; i--) { /* Go forward while *IN* range. */ 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; }

如果弄明白了插入和删除，那么查询就简单多了。这里就不再赘述。

三、总结

跳表因为它的简单，目前应用还是比较广泛的，不光是在Redis中，在区块链的的底层数据库LevelDB和RocksDB以及谷歌的bigtable等，其实在一些语言的底层框架实现中也使用了类似的方法，比如JDK中的ConcurrentSkipListMap等，如果想简单的理解，你就认为是把树打平，树从感官是不断扩散的，层数越多，扩散越明显，遍历越不方便，也不易于理解。特别涉及到旋转等操作。而对于跳表，就易于理解了，说简单一些就如开篇所讲，就是一个二分的多层链表，理论上讲，数据量越大，层数就会越高，控制也会更合理。明白这一点，就非常容易的理解了跳表的原理，只不过跳表的层高不是通过简单的指定来实现的，而是通过概率随机造成的分布稳定性来实现的。 明白这一点，是至关重要的。掌握了原理，再去看源码，就事半功倍了。