redis源码分析与思考（五）——跳跃表

    在计算机科学领域&＃xff0c;跳跃链表是一种数据结构&＃xff0c;允许快速查询一个有序连续元素的数据链表。快速查询是通过维护一个多层次的链表&＃xff0c;且每一层链表中的元素是前一层链表元素的子集。一开始时&＃xff0c;算法在最稀疏的层次进行搜索&＃xff0c;直至需要查找的元素在该层两个相邻的元素中间。这时&＃xff0c;算法将跳转到下一个层次&＃xff0c;重复刚才的搜索&＃xff0c;直到找到需要查找的元素为止。-------维基百科

    简单的来说&＃xff0c;跳跃表是一个有序的数据集合&＃xff0c;里面每个跳跃节点存贮着大量指向其它跳跃节点的指针&＃xff0c;来实现快速访问其它节点的目的。它的性能可以媲美平衡树&＃xff0c;且比平衡树更容易实现。如图所示&＃xff08;图片来自维基百科&＃xff09;&＃xff1a;

       而在Redis中&＃xff0c;跳跃表是作为有序集合的底层实现之一&＃xff0c;当有序集合含有的数据比较多&＃xff0c;又或者数据是比较长的字符串时&＃xff0c;Redis就会采用跳跃表来实现有序集合。

定义

//跳跃表节点
typedef struct zskiplistNode {// 成员对象robj *obj;// 分值double score;// 后退指针struct zskiplistNode *backward;// 层struct zskiplistLevel {// 前进指针struct zskiplistNode *forward;// 跨度unsigned int span;} level[];
} zskiplistNode;


    obj指向的是一个sds&＃xff0c;而level中存贮着跳向另外节点的指针与它们之间的跨度&＃xff0c;跨度指的是从该节点到目标节点跨过的多少节点。 无目标节点则为0&＃xff0c;有目标节点从1开始计算。且跨度是用来计算该节点的排位值的&＃xff0c;在遍历时该节点经历过的节点的跨度和即为该节点的排位值。

//跳跃表
typedef struct zskiplist {// 表头节点和表尾节点struct zskiplistNode *header, *tail;// 表中节点的数量unsigned long length;// 表中层数最大的节点的层数int level;
} zskiplist;


创建

//节点的创建
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {// 分配空间zskiplistNode *zn &＃61; zmalloc(sizeof(*zn)&＃43;level*sizeof(struct zskiplistLevel));// 设置属性zn->score &＃61; score;zn->obj &＃61; obj;return zn;
}//跳跃表的创建
#define ZSKIPLIST_MAXLEVEL 32 //最大的层次&＃xff0c;32
zskiplist *zslCreate(void) {int j;zskiplist *zsl;// 分配空间zsl &＃61; zmalloc(sizeof(*zsl));// 设置高度和起始层数&＃xff0c;默认最少只有一层levelzsl->level &＃61; 1;zsl->length &＃61; 0;zsl->header &＃61; zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);//初始化头结点for (j &＃61; 0; j < ZSKIPLIST_MAXLEVEL; j&＃43;&＃43;) {zsl->header->level[j].forward &＃61; NULL;zsl->header->level[j].span &＃61; 0;}zsl->header->backward &＃61; NULL;// 设置表尾zsl->tail &＃61; NULL;return zsl;
}


    需注意的是&＃xff0c;在Redis中节点的层次大小是随机的&＃xff0c;是采用了幂次定律&＃xff0c;即越大的数几率越小&＃xff0c;它会产生一个1至32的数。

#define ZSKIPLIST_P 0.25
int zslRandomLevel(void) {int level &＃61; 1;while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))// 每往上提一层的概率为4分之一level &＃43;&＃61; 1;return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}


释放

//释放节点
void zslFreeNode(zskiplistNode *node) {//释放对象decrRefCount(node->obj);zfree(node);
}
//释放链表
void zslFree(zskiplist *zsl) {zskiplistNode *node &＃61; zsl->header->level[0].forward, *next;// 释放表头zfree(zsl->header);// 释放表中所有节点while(node) {next &＃61; node->level[0].forward;zslFreeNode(node);node &＃61; next;}// 释放跳跃表结构zfree(zsl);
}


    值得注意的是释放节点函数中的decrRefCount函数&＃xff0c;因为Redis自己用c实现了对象与对象的回收机制&＃xff08;计数回收&＃xff09;&＃xff0c;调用这个方法其实就是在调用该对象的数量上减一。

遍历

    跳跃表的遍历是采用前进指针来完成的。而采用后退指针可以完成从分值大的到小的一次遍历。遍历的过程如图所示&＃xff08;黑色表示前进遍历&＃xff0c;红色表示后退遍历&＃xff09;&＃xff1a;

    跳跃表的查找都是每次从每个跳跃节点的最高层次开始查找的&＃xff0c;是跳着查找的&＃xff0c;查找的过程代码所示&＃xff1a;

x &＃61; zsl->header;for (i &＃61; zsl->level-1; i >&＃61; 0; i--) {while (x->level[i].forward &&(x->level[i].forward->score < score ||// 比对分值(x->level[i].forward->score &＃61;&＃61; score &&// 比对对象&＃xff0c;T &＃61; O(N)compareStringObjects(x->level[i].forward->obj,obj) < 0)))// 沿着前进指针移动x &＃61; x->level[i].forward;}


    如图&＃xff1a;

插入

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;unsigned int rank[ZSKIPLIST_MAXLEVEL];int i, level;//确保score有值redisAssert(!isnan(score));// 在各个层查找节点的插入位置x &＃61; zsl->header;for (i &＃61; zsl->level-1; i >&＃61; 0; i--) {//每次从最高层开始找//rank用来记录与新节点相连的跨过的节点数rank[i] &＃61; i &＃61;&＃61; (zsl->level-1) ? 0 : rank[i&＃43;1];// 沿着前进指针遍历跳跃表while (x->level[i].forward &&(x->level[i].forward->score < score ||// 比对分值(x->level[i].forward->score &＃61;&＃61; score &&// 比对成员compareStringObjects(x->level[i].forward->obj,obj) < 0))) {// 记录沿途跨越了多少个节点rank[i] &＃43;&＃61; x->level[i].span;// 移动至下一指针x &＃61; x->level[i].forward;}// 记录要和新节点相连接的节点update[i] &＃61; x;}// 获取一个随机值作为新节点的层数level &＃61; zslRandomLevel();// 如果新节点的层数比表中其他节点的层数都要大// 那么初始化表头节点中未使用的层&＃xff0c;并将它们记录到 update 数组中// 将来也指向新节点if (level > zsl->level) {// 初始化未使用层for (i &＃61; zsl->level; i < level; i&＃43;&＃43;) {rank[i] &＃61; 0;update[i] &＃61; zsl->header;update[i]->level[i].span &＃61; zsl->length;}// 更新表中节点最大层数zsl->level &＃61; level;}// 创建新节点x &＃61; zslCreateNode(level,score,obj);// 将前面记录的指针指向新节点&＃xff0c;并做相应的设置for (i &＃61; 0; i < level; i&＃43;&＃43;) {// 设置新节点的 forward 指针x->level[i].forward &＃61; update[i]->level[i].forward;// 将沿途记录的各个节点的 forward 指针指向新节点update[i]->level[i].forward &＃61; x;// 计算新节点跨越的节点数量x->level[i].span &＃61; update[i]->level[i].span - (rank[0] - rank[i]);// 更新新节点插入之后&＃xff0c;沿途节点的 span 值// 其中的 &＃43;1 计算的是新节点update[i]->level[i].span &＃61; (rank[0] - rank[i]) &＃43; 1;}// 未接触的节点的 span 值也需要增一&＃xff0c;这些节点直接从表头指向新节点for (i &＃61; level; i < zsl->level; i&＃43;&＃43;) {update[i]->level[i].span&＃43;&＃43;;}// 设置新节点的后退指针x->backward &＃61; (update[0] &＃61;&＃61; zsl->header) ? NULL : update[0];if (x->level[0].forward)x->level[0].forward->backward &＃61; x;elsezsl->tail &＃61; x;// 跳跃表的节点计数增一zsl->length&＃43;&＃43;;return x;
}


    这段代码比较的难懂&＃xff0c;简单的解释一下&＃xff0c;其中有两个关键的数组&＃xff0c;一个是update&＃xff0c;另外一个是rank。update记载的是与新节点有关联的节点&＃xff0c;而rank记载的是与update数组中相对应着的rank值&＃xff0c;而且rank[0]表示着新节点的前驱节点的排位值&＃xff0c;所以它是计算排位值与span值的关键。

删除

void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {int i;// 更新所有和被删除节点 x 有关的节点的指针&＃xff0c;解除它们之间的关系for (i &＃61; 0; i < zsl->level; i&＃43;&＃43;) {//如果该节点的前进节点是目标节点if (update[i]->level[i].forward &＃61;&＃61; x) {update[i]->level[i].span &＃43;&＃61; x->level[i].span - 1;update[i]->level[i].forward &＃61; x->level[i].forward;//如果不是&＃xff0c;将span值减一} else {update[i]->level[i].span -&＃61; 1;}}// 更新被删除节点 x 的前进和后退指针if (x->level[0].forward) {x->level[0].forward->backward &＃61; x->backward;} else {zsl->tail &＃61; x->backward;}// 更新跳跃表最大层数&＃xff08;只在被删除节点是跳跃表中最高的节点时才执行&＃xff09;//因为此时最高层的节点已不存在while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward &＃61;&＃61; NULL)zsl->level--;// 跳跃表节点计数器减一zsl->length--;
}//删除指定分数的、指定对象的节点
int zslDelete(zskiplist *zsl, double score, robj *obj) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;int i;// 遍历跳跃表&＃xff0c;查找目标节点&＃xff0c;并记录所有沿途节点x &＃61; zsl->header;for (i &＃61; zsl->level-1; i >&＃61; 0; i--) {// 遍历跳跃表的复杂度为 while (x->level[i].forward &&(x->level[i].forward->score < score ||// 比分值(x->level[i].forward->score &＃61;&＃61; score &&// 比对象compareStringObjects(x->level[i].forward->obj,obj) < 0)))// 沿着前进指针移动x &＃61; x->level[i].forward;// 记录与指定删除节点有关联的节点update[i] &＃61; x;}//检查找到的元素 x &＃xff0c;只有在它的分值和对象都相同时&＃xff0c;才将它删除。 x &＃61; x->level[0].forward;if (x && score &＃61;&＃61; x->score && equalStringObjects(x->obj,obj)) {zslDeleteNode(zsl, x, update);zslFreeNode(x);return 1;} else {return 0; /* not found */}return 0; /* not found */
}


总结

1. 跳跃表是有序集合的底层实现之一&＃xff0c;每个跳跃表节点的高度是1至32之间的随机数。
2. 跳跃表的大小是随机的&＃xff0c;采用了幂次定律来生成节点的层次大小。
3. 理论上&＃xff0c;节点的层次大小越高&＃xff0c;跳跃表的效率就越高。可以插入分值相同但是对象不同的节点