Redis源码剖析——skiplist的实现
程序员文章站
2024-02-29 12:40:22
...
跳跃表skiplist
跳跃表是一种有序的数据结构,它通过用空间换时间,在每个节点中维持多个指向其他节点的指针,从而达到快速访问的目的。跳跃表插入、删除的平均复杂度为O(logN),最坏为O(N),可以和红黑树相媲美,但是在实现起来,比红黑树简单很多。
Redis使用跳跃表来实现有序集合对象
基本结构
// 跳跃表节点
typedef struct zskiplistNode {
// 成员对象
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;
用图很形象的表示
跳跃表是有序数据结构,按分值进行排序,score相同的情况下比较字符串对象的大小,level[i]中的forward指针只能指向与它有相同层级的节点
基本函数
函数 | 操作 | 复杂度 |
---|---|---|
zslCreateNode | 创建一个跳跃表节点 | O(1) |
zslCreate | 创建一个跳跃表 | O(1) |
zslInsert | 插入一个跳跃表节点 | 平均O(log N),最坏O(N) |
zslDeleteNode | 根据传入的节点指针删除该节点 | O(1) |
zslDelete | 根据分数删除一个节点 | 平均O(log N),最坏O(N) |
zslCreateNode
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;
}
zslCreate
/*
* 创建一个跳跃表
* 表头节点默认为32层
* /
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;
}
zslInsert
理解了插入操作,后面的所有操作就很容易理解了
最主要的就是update和rank两个数组
代码分析如下:
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
// update存储每一层位于插入节点的前一个节点
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
// rank存储每一层位于插入节点的前一个节点的排名
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
redisAssert(!isnan(score));
// 在各个层查找节点的插入位置
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 如果 i 不是 zsl->level-1 层
// 那么 i 层的起始 rank 值为 i+1 层的 rank 值
// 各个层的 rank 值一层层累积
// 最终 rank[0] 的值加一就是新节点的前置节点的排位
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;
}
// 获取一个随机值作为新节点的层数
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;
}
// 创建新节点
x = zslCreateNode(level,score,obj);
// 将前面记录的指针指向新节点,并做相应的设置
for (i = 0; i < level; i++) {
// 设置新节点的 forward 指针
x->level[i].forward = update[i]->level[i].forward;
// 将沿途记录的各个节点的 forward 指针指向新节点
update[i]->level[i].forward = x;
// 计算新节点跨越的节点数量
x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
// 更新新节点插入之后,沿途节点的 span 值
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
// 高层节点跨越了新节点,跨度加1
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;
}
大致步骤如下:
1. update数组存储每一层位于插入节点前的节点
2. rank数组存储每一层位于插入节点的前一个节点的排名
3. 创建节点X,调整X的level数组,调整update数组中节点的span,forward
4. 设置新节点后退指针,zsl长度加1,返回
注:排名rank指的就是当前节点是第几个节点
zslDeleteNode
有了insert中对update数组的理解,这个函数就很容易了
改update数组中节点的span,forward和节点X下一个节点的backward指针
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
// 更新所有和被删除节点 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;
}
}
// 更新被删除节点 x 的前进和后退指针
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--;
}
zslDelete
根据分数和对象删除节点
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;
}
//检查找到的元素 x ,只有在它的分值和对象都相同时,才将它删除。
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 */
}
return 0; /* not found */
}
小节
跳表用于实现有序集合对象,通过在节点中放入多个指针,一步跨越多个节点,空间换时间,使查找和插入的平均时间为O(log N)。实现起来也比较简单,容易理解,没红黑树那么复杂。