4. Redis的有序集合底层---跳跃表
1. 跳跃表简介
示例1.png跳跃表是一种有序的数据结构,它通过在每个节点上维持多个指向其它节点的指针来达到快速访问的目的。跳跃表在插入、删除和查找操作上的平均复杂度为
O(logN)
,最坏为O(N)
,可以和红黑树相媲美,但是在实现起来,比红黑树简单很多。在Redis中的zset
就是使用了跳跃表的实现有序集合。
如上图所示,有一个链表保存着若干有序的数字,如果我们想找到
<16,23,25,31>
这几个数字的话,简单的方法就是遍历一遍,时间复杂度为O(N)
。数据量小的话还好,但是数据量一旦十分庞大的话就非常耗时了。这个时候就是需要我们去优化的了,找一个线性的顺序表,我们一般使用二分法查找,时间复杂度为O(LogN)
。但是链表不适用二分法去查找(链表没有线性表那种连续下标,无法确定中间位置)。于是跳表产生了,它是一种随机化的数据结构,类似二叉搜索树,我们把一些节点提取出来作为索引。如下:提取一级索引.png
我们提取了<13,19,25,34>作为一级索引,这样可以减少比较的次数。同样的,有一级索引就可以建立二级索引,如下,在海量的数据情况下这种方式无疑会大大减少查询的时间
O(logN)
二级索引.png
跳跃表具有以下性质:
a. 跳跃表是由很多层结构组成,并且每一层都是一个有序的链表
b. 最底层(Level 1)的链表包含所有元素
c. 跳跃表(Level 2)开始每个节点包含2个指针,一个指向同一链表的下一个元素,一个指向下一层的元素,如上图。
2. Redis中跳跃表的结构(C语言实现)
在上面简单了解了跳跃表的基本含义及实现思想。在redis中,跳跃表的定义在<server.h>
中,包含2个结构体,分别表示跳跃表节点的zskiplistNode和跳跃表zskilist。zskiplistNode 结构用于表示跳跃表节点,zskiplist结构用来保存跳跃表节点的相关信息。源码如下:
typedef struct zskiplistNode {
//节点对应成员的对象,一般是SDS
robj *obj;
//分数,跳跃表的顺序按照分值进行排列
double score;
//存储后继节点的指针
struct zskiplistNode *backward;
// 层级
struct zskiplistLevel {
// 存储前驱节点的指针
struct zskiplistNode *forward;
// 跨度
unsigned int span;
} level[];
} zskiplistNode;
typedef struct zskiplist {
// 跳跃表的表头节点和表尾节点
struct zskiplistNode *header, *tail;
// 跳跃表中节点的数量
unsigned long length;
// 表中层数最大的节点层数
int level;
} zskiplist;
level[]
是跳跃表中非常重要的属性,在初始化一个跳跃表节点的时候会为其随机生成一个层级大小,这个大小通常在1-32之间(不需要太大,假设极端平均的情况下2^32已经是很大了42亿+
),作为level数组的大小。数组中的每个元素包含一个指向其它节点的前进指针,通过它来加速访问其它节点。如下图所示,这个层级是随机产生的,所以每个节点拥有的层级是不一定的,但是至少是1级。假设链表<13,16,19,34>
它的跳跃表结构可能
如下(注意这里说的是可能,因为每个节点的层级Level
是随机的):
说明:箭头上的1,2等为span的值也就是跨度,k13代表13这个score的key。虚线箭头代表从表尾到表头的链,在从后往前取值的场景中适用的。
上面对跳跃表的结构做了一个介绍,这个时候基本对跳跃表有一定的认识了,下面再来看下跳跃表的插入
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* 到达插入位置所需要经过的level层级*/
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 &&
compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
在跳跃表中的查找不需要从头遍历到表尾。而是像上面的for一样,从头节点开始,如果level数组的forward指针指向的节点的score值大于要插入的score,那么就下降一层;否则,就把x前进一个节点,指向到下一个节点,继续比较,即上述代码中while循环所做的工作。最后,当结束for循环时,update数组中保存的节点的forward就是将要插入的节点的level数组中的forward需要的值,并且待插入的节点一定是位于update[0]这个指针所指节点的后边。
当新节点需要插入的位置找到后,就需要确定新节点的层数:
level = zslRandomLevel();
我们知道,跳跃列表是对有序的链表增加上附加的前进链接,增加是以随机化的方式进行的,所以节点层数的确定当然也是随机的,以上这行代码便是确定节点的层数。
因为在前边查找的时候,是从当前跳表的最高一层开始查找的,如果新节点的层数大于跳表当前的层数,则需要更新跳表的层数并扩展之前的update数组,代码如下:
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;
}
这里需要指出的是为什么新加的层节点直接用zsl->header赋值呢? 原因是这样的,因为新加入的这层与zsl->header直接肯定是没有其他节点的层的;而下边我们在给初始化新插入节点的level数组的时候,是把update的每一个元素当作其前一个跳跃节点的。
x = zslCreateNode(level,score,obj);
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;
}
综合这几段代码,我们便可以知道update这个数组的作用了;然后便是给backward赋值,以及修改zsl的tail指针:
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++;
那么我们再来看下跳跃表如何删除元素:
同插入节点类似,删除节点的时候,同样也是需要先找到需要删除节点的位置,代码如下:
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 &&
compareStringObjects(x->level[i].forward->obj,obj) < 0)))
x = x->level[i].forward;
update[i] = x;
}
这里,需要特别注意的是,到for循环结束时,update[0]处的指针所指向的是要删除的节点的前一个节点,到这里为止,要删除的节点是否存在并不知道。
x = x->level[0].forward;
这里便是将节点往后前进一个位置,便到了我们要寻找的节点的位置,到这里为止,要删除的节点是否存在仍然不知道,下边便是做节点的比较:
if (x && score == x->score && equalStringObjects(x->obj,obj)) {
zslDeleteNode(zsl, x, update);
zslFreeNode(x);
return 1;
} else {
return 0; /* not found */
}
只有当score值与节点元素的值全部相等时,才说明要删除的节点是存在的,否则就是不存在的。
最后,节点元素的查找跟之前插入跟删除节点的查找过程是一样的,具体请看redis的实现 t_zset.c unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) 这个函数。