Redis源码剖析–跳跃表
目录
跳跃表(skiplist)是一种有序数据结构, 它通过在每个节点中维持多个指向其他节点的指针, 从而达到快速访问节点的目的。
跳跃表支持平均 O(\log N) 最坏 O(N) 复杂度的节点查找, 还可以通过顺序性操作来批量处理节点。
在大部分情况下, 跳跃表的效率可以和平衡树相媲美, 并且因为跳跃表的实现比平衡树要来得更为简单, 所以有不少程序都使用跳跃表来代替平衡树。
Redis 使用跳跃表作为有序集合键的底层实现之一: 如果一个有序集合包含的元素数量比较多, 又或者有序集合中元素的成员(member)是比较长的字符串时, Redis 就会使用跳跃表来作为有序集合键的底层实现。
和链表、字典等数据结构被广泛地应用在 Redis 内部不同, Redis 只在两个地方用到了跳跃表, 一个是实现有序集合键, 另一个是在集群节点中用作内部数据结构, 除此之外, 跳跃表在 Redis 里面没有其他用途。
先看一下维基百科对跳跃表的图示:
从图中可以看到, 跳跃表主要由以下部分构成:
- 表头(head):负责维护跳跃表的节点指针。
- 跳跃表节点:保存着元素值,以及多个层。
- 层:保存着指向其他元素的指针。高层的指针越过的元素数量大于等于低层的指针,为了提高查找的效率,程序总是从高层先开始访问,然后随着元素值范围的缩小,慢慢降低层次。
- 表尾:全部由 NULL 组成,表示跳跃表的末尾。
跳跃表结构定义
跳跃表的结构体定义在server.h文件中。其中包括跳跃表节点zskiplistNode和跳跃表zskiplist两个结构体。
/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
sds ele; // 具体成员对象
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;
每次创建一个新跳跃表节点的时候, 程序都根据幂次定律 (power law,越大的数出现的概率越小) 随机生成一个介于 1 和 32 之间的值作为 level 数组的大小, 这个大小就是层的“高度”。
总的层数保存在zskiplist的level参数中, 另外每个节点保存了各自层中的指针以及这一层的跨度。
跳跃表操作
创建跳跃表
/* Create a skiplist node with the specified number of levels.
* The SDS string 'ele' is referenced by the node after the call. */
// 创建跳跃表结点
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;
}
/* 创建跳跃表 */
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
// 申请内存
zsl = zmalloc(sizeof(*zsl));
// 设置层数为1
zsl->level = 1;
// 总的结点数为0
zsl->length = 0;
// ZSKIPLIST_MAXLEVEL=32, 头结点,设置最大层数,分值为0, 具体对象为NULL
zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
// 循环设置头结点的每一层的前向指针为NULL,跨度为0
for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
zsl->header->level[j].forward = NULL;
zsl->header->level[j].span = 0;
}
// 头结点的后向指针为NULL
zsl->header->backward = NULL;
zsl->tail = NULL;
return zsl;
}
创建跳跃表的时候设置层数为1, 只有一个头结点,头结点保存了最大的层数,同时所有的前向指针都为NULL。
释放整个跳跃表
/* Free the specified skiplist node. The referenced SDS string representation
* of the element is freed too, unless node->ele is set to NULL before calling
* this function. */
void zslFreeNode(zskiplistNode *node) {
sdsfree(node->ele);
zfree(node);
}
/* 释放整个跳跃表. */
void zslFree(zskiplist *zsl) {
// 从最底层的level[0]依次遍历,释放
zskiplistNode *node = zsl->header->level[0].forward, *next;
zfree(zsl->header);
// 如果存在后续结点
while(node) {
next = node->level[0].forward;
// 释放结点
zslFreeNode(node);
node = next;
}
zfree(zsl);
}
跳跃表插入元素
/* Insert a new node in the skiplist. Assumes the element does not already
* exist (up to the caller to enforce that). The skiplist takes ownership
* of the passed SDS string 'ele'. */
// 跳跃表插入元素
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score)); // 判断是否为数字
x = zsl->header;
// 从最高的level, 也即跨度最大的level开始查找结点
for (i = zsl->level-1; i >= 0; i--) {
/* store rank that is crossed to reach the insert position */
// 当前是否是最高层, 如果是最高层,rank[i]=0,否则,复制上一层的数值
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
// 如果当前结点的score值小于传入的score 或者 当前score相等,但是结点的对象不相等
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[i]
rank[i] += x->level[i].span;
// 在当前层中向前查找
x = x->level[i].forward;
}
// 当前层位于插入位置前的结点x放入update数组
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. */
// 随机生成小于32的层数
level = zslRandomLevel();
// 如果生成的层数大于当前的层数
if (level > zsl->level) {
for (i = zsl->level; i < level; i++) {
// 设定rank数组中大于原level层以上的值为0
// 同时设定update数组大于原level层以上的数据
rank[i] = 0;
update[i] = zsl->header;
update[i]->level[i].span = zsl->length;
}
zsl->level = 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 */
// 更新每一层的前置结点和新结点的跨度
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++;
}
// 根据最低层的前序结点是否是header结点来设置当前新结点的向后指针
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;
}
首先看图示,如果想要插入score=5的结点(redis中允许score值重复),那么首选需要找到score=5的结点,查找的顺序为:
从最大的层向后查找,如果当前层后边没有值了,并且当前结点的值小于要找的值,就查找下一层结点;如果下一个结点的值大于要找的值,也会到下一层结点继续查找。
找到对应的位置执行插入操作后,需要为新结点设置层数,那么设置多少层合适呢,这边直接采用了一个随机数。随机数生成了多少层,当前新结点的层数就设置多少层。
如果新层数小于原来的层数,只需要重新设置前序后置结点的指针和跨度就行;如果新层数大于原来的层数,就需要额外设置新的更高的层。
那么插入结点之后,如何修改前序和后置结点的指针和跨度呢。这边用了两个数组来记录,分别是update和rank。update用来记录每一层中插入位置的前序结点,到时候根据这个前序结点设置每一层的指针调整。 rank用来记录每一层到新结点的跨度,插入新结点之后,根据rank数组中记录跨度更新前置结点的跨度值。
删除跳跃表结点
/* 删除结点,如果结点存在并删除,返回1, 否则返回0
参数中node如果是空的,则结点确实被删除; 如果非空,只是把结点从链表上摘下来,返回指针给node*/
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. */
// 由于允许存在相同的score,需要在score和ele都满足的条件下才能删除
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 */
}
具体的结点删除操作:
/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
// update数组保存了所有层上要删除结点的前置结点
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
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 {
// 否则,只需要将前置结点的跨度减1即可,因为少了一个结点啊
update[i]->level[i].span -= 1;
}
}
// 修改backward指针,需要考虑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--;
}