Redis源码剖析——跳跃表

Redis使用跳跃表作为有序集合键的底层实现之一,跳跃表(skiplist)是一种有序数据结构,它通过在每个节点中维持多个指向其他节点的指针,从而达到快速访问节点的目的。

跳跃表支持平均O(logN)、最坏O(N)复杂度的节点查找,还可以通过顺序性操作来批量处理节点。在大部分情况下,跳跃表的效率可以和平衡树相媲美,并且因为跳跃表的实现比平衡树要来得更为简单,所以有不少程序都使用跳跃表来代替平衡树。

数据结构

一个跳跃表的整体结构如下:

跳跃表节点

跳跃表节点的数据结构定义如下:

1
2
3
4
5
6
7
8
9
10
11
// 跳跃表节点
typedef struct zskiplistNode {
robj *obj; // 保存的值
double score; // 节点分值
struct zskiplistNode *backward; // 后退指针
// 层
struct zskiplistLevel {
struct zskiplistNode *forward; // 前进指针
unsigned int span; // 跨度
} level[];
} zskiplistNode;

  • 层:level数组包含多个元素,每个元素都包含一个指向其他节点的指针,程序可以通过这些层来加快访问其他节点的速度,一般来说,层的数量越多,访问其他节点的速度就越快
  • 前进指针:每个层都有一个指向表尾方向的前进指针,用于从表头向表尾方向访问节点
  • 跨度:表示两个节点之间的距离(前进指针指向节点和当前节点的距离)
  • 后退指针:用于从表尾向表头访问节点
  • score分值:跳跃表中所有的节点按照分值进行排序
  • obj:保存的成员,一般为sds数据结构

跳跃表

1
2
3
4
5
6
// 跳跃表
typedef struct zskiplist {
struct zskiplistNode *header, *tail; // 头结点、尾节点
unsigned long length; // 长度
int level; // 最高层节点的层数
} zskiplist;

跳跃表记录了头结点和尾结点的指针、长度(即跳跃表中节点数目)和层数最大的节点的层数,注意表头节点的层高并不计算在内。

基本操作

创建跳跃表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// 创建跳跃表节点
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
// 分配内存空间
zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
// 初始化
zn->score = score;
zn->obj = obj;
return zn;
}
// 创建跳跃表
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
// 分配内存空间
zsl = zmalloc(sizeof(*zsl));
zsl->level = 1;
zsl->length = 0;
// 表头为32层的空节点,每一层都指向NULL
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;
}

可见:

  • 创建的跳跃表的初识level层数值为1
  • 刚创建的跳跃表的头结点是一个有32层的空节点,其中每一层的forward都是NULL

跳跃表插入节点

跳跃表插入节点的部分有些复杂,需要改变节点前后节点的forward、backward指针以及长度等信息。其基本思想是:使用update表记录新节点在各层中forward指针指向它的节点,然后插入,同时改变这些复杂的指向关系。

其中rank数组是用来记录每一个节点再整个节点表中的排位信息,其是通过每层中的跨度计算得来的。

新插入的节点的层数是通过幂次定律决定的一个1-32的数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
// 返回一个1-32的层数值,使用幂次定律,越大的数出现的概率越小
int zslRandomLevel(void) {
int level = 1;
// 使用幂次定律,1/4概率
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
// 跳跃表插入节点
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
// 记录每一个节点的排位信息
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
redisAssert(!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 &&
// 对比成员
compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
// 计算排位
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
// 第i层将要和新节点连接的节点
update[i] = x;
}
/* we assume the key is not already inside, since we allow duplicated
* scores, and the re-insertion of score and redis object 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,obj);
// 使用update中的信息将为新的节点建立连接
for (i = 0; i < level; i++) {
// 设置新节点的forward
x->level[i].forward = update[i]->level[i].forward;
// 前面的节点的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;
}

跳跃表删除节点

删除节点的方法差不多就是插入节点方法的反向操作,首先找到目标节点(通过update数组记录沿途节点),接触forward指针关系,更新跳跃表的层数和长度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
// 跳跃表内部删除节点,update数组为forward指向本节点的数组
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
// 解除forward关系
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;
}
}
// 更新backward指针
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--;
}
/* Delete an element with matching score/object from the skiplist. */
// 跳跃表删除节点
int zslDelete(zskiplist *zsl, double score, robj *obj) {
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;
}
/* 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 && equalStringObjects(x->obj,obj)) {
zslDeleteNode(zsl, x, update);
zslFreeNode(x);
return 1;
}
return 0; /* not found */
}

跳跃表查找节点

跳跃表中查找节点相关的操作主要有获取排名、依据排名获取信息。其基本思想和插入节点找到插入位置一样。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
// 通过分值和对象值获取排位信息,以1为起始值
unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
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 &&
compareStringObjects(x->level[i].forward->obj,o) <= 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->obj && equalStringObjects(x->obj,o)) {
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;
}

用一张图能够简单表示这种查找行为,例如找到5这个节点:

56782