redis设计与实现——跳表

发表于

redis设计与实现——跳表

skiplist跳跃表是一种有序数据结构，通过在每个节点中维持多个节点中维持多个指向其他节点的指针。节点查找的时间复杂度，平均是O(logN)，最坏是O(N)，还可以批量操作处理节点。关于skiplist，可以参考论文。

在Redis中，跳跃表是作为有序集合键的底层实现基础。

跳跃表的实现

Redis的跳跃表是在server.h/zskiplistNode和server.h/zskiplist两个结构体中定义的，其中前者表示表节点，后者用来保存节点的数量，头部和尾部节点指针等相关信息。

img

跳跃表节点

跳跃表节点的结构体定义为：

1
2
3
4
5
6
7
8
9
10
/* 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;
  • level：这是跳跃表节点的level数组，包含了多个元素，每个元素都不包含了一个指向前方其他节点的指针。span则是跨度，如上图所示，箭头上方的就是跨度数值，在查找某个节点的过程中，将沿途访过的所有层的跨度累积起来，就是当前节点所在的排位；
  • backward：后退指针，只有一个后退指针意味着每次只能往前一个节点；
  • score：跳跃表中所有节点按照分数值从小到大排序；
  • ele：指向一个SDS字符对象，ele是唯一的，因此score相同时，需要按照sds在字典序中的大小排序；

跳跃表

跳跃表的结构体定义为：

1
2
3
4
5
typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

redis使用zskiplist来管理所有的节点，比如表头节点和表尾节点，跳跃表长度以及level则表示在O(1)复杂度内获取跳跃表中层高最大的那个节点的层数量。

另外表头节点虽然与其他节点构造一样，但其后退指针、分值和ele等属性都不使用，并且不参与zskiplist中level的计算。

创建跳跃表

创建跳跃表的操作主要是完成一些初始化的操作，其时间复杂度为O(1)。创建操作主要依赖两个函数：zslCreate()和zslCreateNode()。其中redis默认跳跃表的最大层级为64。

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
/* Should be enough for 2^64 elements */
#define ZSKIPLIST_MAXLEVEL 64 // zskiplist的最大层级为64

/* 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);//创建一个节点
  	// 表头指针的后退指针和分值，ele都不作使用
    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;
}

/* 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;
}

插入新节点

新节点的插入是通过zslInsert实现的，给定分数值和ele元素则可返回新的跳跃表节点。

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
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    // 记录每一层插入节点的前面一个节点在skiplist中的排名
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!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];
      	// 先根据score比较，相等即根据sds的字符串字典序比较
        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;// 每一个层级的待插入位置，即当前层最后一个小于x的点
    }
    /* 重复插入相同的元素，这种情况是不会发生的
     * 如果元素在内部，则zslInsert()的调用者应该在哈希表中进行测试是否在里面 */
    level = zslRandomLevel();
  	// 如果计算出来的层级比当前层级高，则重设超出zsl原来层级的指针
    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,ele); // 创建新的节点
    for (i = 0; i < level; i++) {
      	// 插入到当前位置(update[i]的前面)
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;
				
      	/* rank[0] - rank[i] 是x在第i层的前一个节点与x之间的距离*/
        /* 更新插入点的跨度 */
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
      	/* 更新插入点前一个节点的跨度 */
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    /* 自增没有到达的层级的span */
    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;
}

其中随机层级的实现如下，redis的跳跃表最大层数为64，能够足够支撑优化2^64个元素的查找。其中获取随机层级时，越高的层级数出现的几率越小，而且每往上一个层级，其概率为1/4。

1
2
3
4
5
6
7
8
9
10
11
12
/* Returns a random level for the new skiplist node we are going to create.
 * The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL
 * (both inclusive), with a powerlaw-alike distribution where higher
 * levels are less likely to be returned. */
int zslRandomLevel(void) {
    int level = 1;
  	// #define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
  	// random()&0xFFFF的结果就是一个0-65535的数
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

总结一下

  • 跳跃表时有序集合的底层实现之一；
  • Redis跳跃表时由在server.h的zskiplistNode和zskiplist实现的；
  • 每个跳跃表节点层高都是1-64的随机数；
  • 多个节点可以包含相同的score，但ele时唯一的；

源码版本

截至2019/07/27，5.0之后的版本，commitId为：505a855000ef8f1fbea9cb41841fa8708175bba4