Redis中的跳表（skiplist）是一种有序的数据结构，它通过在每个节点中维护多个指向其他节点的指针，从而实现快速访问。跳表在插入、删除和查找操作上的平均复杂度为O(logN)，最坏情况下为O(N)，与红黑树相媲美，但实现起来比红黑树简单很多[1]。
跳表的数据结构定义在Redis的server.h文件中，包括跳表节点（zskiplistNode）和跳表（zskiplist）两个结构体。跳表节点包含成员对象、分值、后向指针和层等属性；而跳表包含表头节点、表尾节点、节点数量和最大层数等属性[1]。
Redis中关于跳表的相关操作函数定义在t_zset.c文件中，下面分别介绍几个基本操作函数的实现源码。
创建跳表：在创建跳表时，需要完成以下操作：
插入节点：往跳表中插入一个节点会改变跳表的长度，可能会改变其层数。插入节点的关键在于找到插入位置，跳表按照分值进行排序，查找步骤大致如下：
删除节点：从跳表中删除一个节点需要完成以下操作：
查找节点：跳表的查找操作与插入操作类似，根据分值进行比较，从最高层开始向前查找，直到找到目标节点或者到达第一层为止[1]。
跳表在Redis中主要用于有序集合（sorted set）的实现，有序集合中的成员对象按照分值进行排序。跳表的高效性和简单实现使得它成为Redis中有序集合的重要数据结构之一。
c/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
sds ele;
double score;
struct zskiplistNode *backward;
struct zskiplistLevel {
struct zskiplistNode *forward;
unsigned long span;
} level[];
} zskiplistNode;
typedef struct zskiplist {
struct zskiplistNode *header, *tail;
unsigned long length;
int level;
} zskiplist;
typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;
接下来是创建跳表的函数
c/* Create a new skiplist. */
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;
}
/* Free a whole skiplist. */
void zslFree(zskiplist *zsl) {
zskiplistNode *node = zsl->header->level[0].forward, *next;
zfree(zsl->header);
while(node) {
next = node->level[0].forward;
zslFreeNode(node);
node = next;
}
zfree(zsl);
}
void zslFreeNode(zskiplistNode *node) {
sdsfree(node->ele);
zfree(node);
}
函数接受一个指向跳表（zskiplist）的指针、一个分数（score）和一个SDS字符串（ele）作为参数，然后在跳表中插入一个新节点。
update是一个数组，用于存储每个层级中插入位置的前一个节点。
rank是一个数组，用于存储到达插入位置所经过的排名。
通过遍历跳表的每个层级，找到合适的插入位置。在遍历过程中，根据节点的分数和字符串值进行比较，以确定插入位置。
如果新节点的层级大于跳表的当前层级，需要更新update和rank数组，并将跳表的层级增加到新节点的层级。
创建新节点并将其插入到跳表中。在插入过程中，更新指针和跨度信息。
最后，根据插入位置更新新节点的backward指针和跳表的尾节点，并增加跳表的长度。
函数返回指向新节点的指针，即插入的节点。
czskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score)); // 确保score值不是NaN（非数字）
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 存储到达插入位置所经过的排名
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 &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
// 在当前层级向前遍历，找到插入位置
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
// 确保元素不存在，因为允许相同分数的元素存在，所以不应该重复插入相同元素
level = zslRandomLevel();
if (level > zsl->level) {
// 如果新节点的层级大于跳表的当前层级，则需要更新update和rank数组
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, ele);
for (i = 0; i < level; i++) {
// 更新指针和跨度
x->level[i].forward = update[i]->level[i].forward;
update[i]->level[i].forward = x;
x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
// 更新未受影响的层级的跨度
for (i = level; i < zsl->level; i++) {
update[i]->level[i].span++;
}
// 设置新节点的backward指针和更新尾节点
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;
}
zslDeleteNode是一个内部函数，被zslDelete、zslDeleteRangeByScore和zslDeleteRangeByRank函数调用。
函数的作用是从跳表中删除指定的节点x。update是一个数组，存储了每个层级中节点x的前一个节点。
函数首先遍历跳表的每个层级，更新前一个节点的指针和跨度信息。如果前一个节点的forward指针指向节点x，则将前一个节点的跨度增加x的跨度减去1，并将前一个节点的forward指针指向x的forward指针；否则，将前一个节点的跨度减去1。
函数更新节点x的前进指针和后退指针。如果节点x的最底层的forward指针不为空，则将其后继节点的backward指针指向x的前驱节点；如果节点x的最底层的forward指针为空，则更新跳表的尾节点为x的前驱节点。
最后，函数检查跳表的层级。如果跳表的层级大于1，并且最高层级的forward指针为NULL（即没有节点存在于该层级），则将跳表的层级减少1。
函数结束后，更新跳表的长度，即将跳表的节点数减1。
c/* Internal function used by zslDelete, zslDeleteRangeByScore and
* zslDeleteRangeByRank. */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
// 从跳表中删除节点x，update是一个数组，存储了每个层级中节点x的前一个节点。
for (i = 0; i < zsl->level; i++) {
// 遍历每个层级，更新前一个节点的指针和跨度
if (update[i]->level[i].forward == x) {
// 如果前一个节点的forward指针指向x，则将跨度增加x的跨度-1，并将前一个节点的forward指针指向x的forward指针
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;
}
}
// 更新x的前进指针和后退指针
if (x->level[0].forward) {
x->level[0].forward->backward = x->backward;
} else {
zsl->tail = x->backward;
}
// 如果跳表的层级大于1，并且最高层级的forward指针为NULL，将跳表的层级减少1
while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
zsl->level--;
// 更新跳表的长度
zsl->length--;
}
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. */
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 */
}
unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec *range, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
!zslValueGteMin(x->level[i].forward->score, range))
x = x->level[i].forward;
update[i] = x;
}
/* Current node is the last with score < or <= min. */
x = x->level[0].forward;
/* Delete nodes while in range. */
while (x && zslValueLteMax(x->score, range)) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x); /* Here is where x->ele is actually released. */
removed++;
x = next;
}
return removed;
}
unsigned long zslDeleteRangeByLex(zskiplist *zsl, zlexrangespec *range, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward &&
!zslLexValueGteMin(x->level[i].forward->ele,range))
x = x->level[i].forward;
update[i] = x;
}
/* Current node is the last with score < or <= min. */
x = x->level[0].forward;
/* Delete nodes while in range. */
while (x && zslLexValueLteMax(x->ele,range)) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x); /* Here is where x->ele is actually released. */
removed++;
x = next;
}
return removed;
}
/* Delete all the elements with rank between start and end from the skiplist.
* Start and end are inclusive. Note that start and end need to be 1-based */
unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned long traversed = 0, removed = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (x->level[i].forward && (traversed + x->level[i].span) < start) {
traversed += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
traversed++;
x = x->level[0].forward;
while (x && traversed <= end) {
zskiplistNode *next = x->level[0].forward;
zslDeleteNode(zsl,x,update);
dictDelete(dict,x->ele);
zslFreeNode(x);
removed++;
traversed++;
x = next;
}
return removed;
}
zslUpdateScore函数用于更新跳表中指定元素的分值。
函数首先需要定位到要更新分值的元素，从跳表的头节点开始查找。通过遍历每个层级，在每个层级中查找分值小于给定分值（curscore）或者分值相同但字典序小于给定元素（ele）的节点。在查找过程中，使用update数组保存每个层级中节点的前一个节点。
定位到要更新分值的元素后，函数进行断言检查，确保该元素的分值和元素值与给定的分值和元素值相同。
如果更新分值后的节点在跳表中的位置没有发生变化，即新的分值仍然使节点保持在原位置，则直接更新节点的分值，无需删除和重新插入节点。
如果无法重用旧节点，即更新分值后的节点需要在跳表中移动到不同的位置，函数会调用zslDeleteNode函数删除旧节点，并调用zslInsert函数在新的位置插入一个新节点。
最后，函数释放旧节点的ele字符串，并释放旧节点的内存空间。
函数返回更新后的节点指针，如果节点在新位置上这个函数的作用是更新跳表中指定元素的分值。
czskiplistNode *zslUpdateScore(zskiplist *zsl, double curscore, sds ele, double newscore) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
int i;
/* We need to seek to element to update to start: this is useful anyway,
* we'll have to update or remove it. */
// 首先需要定位到需要更新分值的元素，从跳表的头节点开始查找
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 在每个层级中，遍历查找当前层级中分值小于给定分值（curscore）或者分值相同但字典序小于给定元素（ele）的节点
while (x->level[i].forward &&
(x->level[i].forward->score < curscore ||
(x->level[i].forward->score == curscore &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
x = x->level[i].forward;
}
update[i] = x;
}
/* Jump to our element: note that this function assumes that the
* element with the matching score exists. */
// 定位到需要更新分值的元素
x = x->level[0].forward;
serverAssert(x && curscore == x->score && sdscmp(x->ele,ele) == 0);
/* If the node, after the score update, would be still exactly
* at the same position, we can just update the score without
* actually removing and re-inserting the element in the skiplist. */
// 如果更新分值后的节点在跳表中的位置没有发生变化，则直接更新分值，不需要删除和重新插入节点
if ((x->backward == NULL || x->backward->score < newscore) &&
(x->level[0].forward == NULL || x->level[0].forward->score > newscore))
{
x->score = newscore;
return x;
}
/* No way to reuse the old node: we need to remove and insert a new
* one at a different place. */
// 无法重用旧节点，需要在跳表中删除旧节点，并在新的位置插入一个新节点
zslDeleteNode(zsl, x, update);
zskiplistNode *newnode = zslInsert(zsl,newscore,x->ele);
/* We reused the old node x->ele SDS string, free the node now
* since zslInsert created a new one. */
x->ele = NULL;
zslFreeNode(x);
return newnode;
}
zslGetRank函数用于获取指定元素在跳表中的排名（即在有序集合中的位置）。
函数从跳表的头节点开始，通过从高层级向低层级遍历，逐层查找节点，直到找到指定元素或找到元素应该插入的位置。
在每个层级中，函数通过比较节点的分值和元素值，决定是继续向前遍历还是停止遍历。如果节点的分值小于给定分值（score），或者分值相同但字典序小于等于给定元素（ele），则继续遍历。
在遍历过程中，函数累加每个遍历过的节点的跨度（span）到rank变量中。跨度表示节点到下一个节点的距离，用于计算排名。
如果找到了指定元素，函数返回当前的排名（即累加的跨度值）。注意，x可能等于zsl->header，所以在判断时需要检查元素指针ele是否非空。
如果没有找到指定元素，函数返回0，表示元素不在跳表中。
cunsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {
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 &&
sdscmp(x->level[i].forward->ele,ele) <= 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->ele && x->score == score && sdscmp(x->ele,ele) == 0) {
return rank;
}
}
return 0;
}
Learn more:
本文作者:yowayimono
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！