折腾Redis之跳跃表

Author：带翅膀的猫
发布时间：September 11, 2020
4635views
No comments
8151 words
Categories： Redis

跳跃表

跳跃表是一种有序的数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而快速访问节点的目的。跳跃表支持平均O(logN)、最坏O(N)复杂度的节点查找，还可以通过顺序性操作来批量处理节点。在大部分情况下，跳跃表的效率可以和平衡树相媲美，并且因为跳跃表的实现比平衡树要来得更为简单，所以有不少程序都使用跳跃表来代替平衡树。

Redis使用跳跃表作为有序集合键的底层实现之一，如果一个有序集合包含的元素数量比较多，又或者有序集合中元素的成员是比较长的字符串是，Redis就会使用跳跃表来作为有序集合键的底层实现。

和链表、字典等数据结构被广泛地应用在Redis内部不同，Redis只在两个地方用到了跳跃表，一个实现有序集合键，另一个是在集群节点中用作内部数据结构，除此之外，跳跃表在Redis中没有其他用途。

如果我们将有序链表的部分节点分层，每一层都是一个有序链表。在查找时优先从最高层开始向后查找。当到达某节点时，如果next节点值大于要查找的值或next指针指向NULL，则从当前节点下降一层继续向后查找。

Redis的跳跃表

Redis的跳跃表由redis.h/zskiplistNode和redis.h/zskiplist两个结构定义，其中zskiplistNode结构用于表示跳跃表节点，而zskiplist结构用于保存跳跃表节点的相关信息，比如节点的数量，以及指向表头节点和表尾节点的指针等等。

请输入图片描述

上图展示了一个Redis跳跃表的示例，位于图片最左边的是zskiplist结构，该结构包含以下属性：

header：指向跳跃表的表头节点
tail：指向跳跃表的表尾节点
level：记录目前跳跃表内层数最大的那个节点的层数(表头节点的层数不计算在内)
length：记录跳跃表长度，即跳跃表目前包含节点的数量(表头不计算在内)

位于zskiplist结构右边的是四个zskiplistNode结构，该结构包含以下属性：

层(level)：节点中用L1,L2,L3等字样标记节点的各个层，L1代表第一层，L2代表第二层，以此类推。每个层都带有两个属性：前进指针和跨度。前进指针用于访问位于表尾方向的其他节点，而跨度则记录了前进指针所指向节点和当前节点的距离。在上图中，连线上带数字的箭头就代表前进指针，而那个数字就是跨度。当程序从表头向表尾进行遍历时，访问会沿着层的前进指针进行。
后退指针：节点中用BW字样标记节点的后腿指针，它指向位于当前节点的前一个节点，后退指针在程序从表尾项表头遍历时使用。
分值：各个节点中的1.0、2.0和3.0就是节点所保存的分值。在跳跃表中，节点按各自保存的分值从小到大排列。
成员对象：各个节点中的o1、o2和o3是节点所保存的成员对象。

注意表头节点和其他节点的构造是一样的：表头节点也有后退指针、分值和成员对象。只是不会用到这些属性而已。

注意这里展示的表头最大层数为32(Redis3.0)，这是根据ZSKIPLIST_MAXLEVEL决定的，在Redis5.x中这个值为64.

节点源码

跳跃表节点定义如下：

typedef struct zskiplistNode {
    //成员对象
    robj *obj;
    //分值
    double score;
    //后退指针
    struct zskiplistNode *backward;
    //层
    struct zskiplistLevel {
        //前进指针
        struct zskiplistNode *forward;
        //跨度
        unsigned int span;
    } level[];
} zskiplistNode;

层

可以发现跳跃表的level数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层来加快访问其他节点的速度，一般来说，层的数量越多访问其他节点的速度就越快。
每次创建一个新的跳跃表节点的时候，程序都根据幂次定律(power law，越大的数出现的概率越小)随机生成一个介于1和ZSKIPLIST_MAXLEVEL之间的值作为level数组的大小，这个大小就是层的高度。

/**
为我们将要创建的zskiplistNode返回一个随机level。这个函数的返回值在1和ZSKIPLIST_MAXLEVEL之间(包括这两个值)，使用类似于幂律的分布，较高的级别不太可能被返回。ZSKIPLIST_P 0.25 */
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

level的初始值为1，通过while循环，每次生成一个随机值，取这个值的低16位作为x，当x小于0.25倍的0xFFFF时，level加1；否则退出while循环。最终返回level和ZSKIPLIST_MAXLEVEL的最小值。

跨度

层的跨度用于记录两个节点之间的距离:

两个节点之间的跨度越大，它们相距越远
指向NULL的所有前进指针跨度都为0，因为它们没有连向任何节点

初步看上去决定跨度有什么用??

跨度实际上是用来计算排位的：在查找某个节点的过程中，将沿途访问过的所有层的跨度累计起来，得到的结果就是目标节点在跳跃表中的排位。

分值和成员

节点的分值(score属性)是一个double类型的浮点数，跳跃表中的所有节点都按分值从小到大来排序。
节点的成员对象(obj属性)是一个指针，它指向一个字符串对象，而字符串对象则保存一个SDS值。

在同一个跳跃表中，各个界定啊保存的成员对象必须是唯一的，但是多个节点保存的分值却可以是相同的：分值相同的节点将按照成员对象在字典序中的大小来进行排序，成员对象较小的节点会排在前面，而成员对象较大的节点则会排在后面。

跳跃表源码

跳跃表定义如下：

typedef struct zskiplist {
    //表头和表尾节点
    struct zskiplistNode *header, *tail;
    //表中节点数量
    unsigned long length;
    //表中最大层数
    int level;
} zskiplist;

使用一个zskiplist结构来持有跳跃表节点，程序可以更方便地对整个跳跃表进行处理。

些许源码阅读

创建节点

//根据传递过来的level、score和obj创建一个节点
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
    //节点的大小=一个节点大小+level*一个层的大小
    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));
    //初始最大level为1
    zsl->level = 1;
    zsl->length = 0;
    //创建头节点，头节点层数为ZSKIPLIST_MAXLEVEL
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    //初始化头节点的所有level
    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;
}

插入节点

不难知道，跳跃表插入节点需要经过以下步骤：

查找要插入的位置
调整跳跃表高度
插入节点
调整后退指针

以插入下方跳跃表为例(level为2，length为3)，插入节点score为31，层高为3。
请输入图片描述

查找位置

查找待插入节点位置的源代码如下所示：

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;
    }
    update[i] = x;
}

重点关注update[ZSKIPLIST_MAXLEVEL]和rank[ZSKIPLIST_MAXLEVEL]这两个数组，特别重要。

update[]：插入节点时，需要更新被插入界定啊每层的前一个节点。由于每层更新的节点不一样，所以将每层需要更新的节点记录在update[i]中。
rank[]：记录当前层从header节点到update[i]节点所经历的跨度，在更新update[i]的跨度和设置新插入节点的跨度时用到。

查找节点的插入位置逻辑如下所示：

初始状态：x = head,i = zsl->level - 1 = 1;
第一次循环：
- rank[1] = 1 == 1 ? 0 : rank[i+1];=>rank[1] = 0;
- x->level[1].forward存在，并且x->level[1].forward.score小于31，说明待插入节点在其后面，进入while循环
- 更新rank，rank[1] += x->level[1].span;=>rank[1] = 1
- x向前进一个节点，成为第一个节点
- 第一个节点的前进指针(x->level[1].forward)为NULL，不进入while循环。update[1]为第一个节点
第二次循环：
- i=0，x为第一个节点
- i与zsl-level-1不相等，rank[0]=rank[0+1]=1
- x->level[0].forward存在，并且x->level[0].forward->score小于31，进入while循环
- 更新rank，rank[0] += x->level[0].span;=>rank[0] = 2
- x向前进一个节点，成为第二个节点
- x->level[0].forward存在，但是x->level[0].forward->score大于41，不再进入while循环。update[0]为第二个节点。

请输入图片描述

调整跳跃表高度

如果插入节点的高度(假设为3)大于跳跃表的高度，我们需要调整跳跃表的高度。

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);//创建带插入节点
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;
}

请输入图片描述

调整backward

x->backward = (update[0] == zsl->header) ? NULL : update[0];//x插入到第一个的话backward为NULL
if (x->level[0].forward)
    x->level[0].forward->backward = x;//下一个节点的backward指针指向自己
else
    zsl->tail = x;//x插入到最后了，需要更新tail
zsl->length++;
return x;

请输入图片描述

跳跃表的应用

在Redis中，跳跃表主要应用于有序集合的底层实现(有序集合的另一种实现方式为压缩链表)。
Redis的配置文件中关于有序集合底层实现的两个配置：

# 与哈希和列表类似，排序集也经过特殊编码，以节省大量空间。
# 这种编码只在排序集的长度和元素低于以下限制时使用:
zset-max-ziplist-entries 128 # 采用压缩列表时，元素个数最大值
zset-max-ziplist-value 64    # 采用压缩列表时，每个元素的字符串长度最大值

采用哪种底层实现在t_zset.c的zaddGenericCommand函数中说明了：

//zset-max-ziplist-entries的值是否为0
//zset-max-ziplist-value小于要插入元素的字符串长度
if (server.zset_max_ziplist_entries == 0 ||
`   server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
{
    zobj = createZsetObject(); //使用跳跃表实现
} else {
    zobj = createZsetZiplistObject(); //使用压缩列表实现
}

一般情况下，不会将zset-max-ziplist-entries配置为0，元素的字符串长度也不会变长，所以在创建有序集合时，默认使用压缩列表的底层实现。zset新插入元素时，会判断以下两种条件：

zset中元素个数大于zset-max-ziplist-entries
插入元素的字符串长度大于zset-max-ziplist-value

一旦满足以上任意一条件则会将压缩列表转为跳跃表：

if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
    zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
    zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);

值得注意的是，zset在转为跳跃表后，即使元素被逐渐删除，也不会重新转为压缩列表。

版权属于：带翅膀的猫
本文链接：https://www.chengpengper.cn/archives/111/
转载时须注明出处及本声明

Last modification：September 13th, 2020 at 12:51 pm

如果觉得我的文章对你有用，请随意赞赏

折腾Redis之跳跃表

带翅膀的猫 • 2020 年 09 月 11 日

跳跃表

Redis的跳跃表

请输入图片描述

上图展示了一个Redis跳跃表的示例，位于图片最左边的是zskiplist结构，该结构包含以下属性：

header：指向跳跃表的表头节点
tail：指向跳跃表的表尾节点
level：记录目前跳跃表内层数最大的那个节点的层数(表头节点的层数不计算在内)
length：记录跳跃表长度，即跳跃表目前包含节点的数量(表头不计算在内)

位于zskiplist结构右边的是四个zskiplistNode结构，该结构包含以下属性：

层(level)：节点中用L1,L2,L3等字样标记节点的各个层，L1代表第一层，L2代表第二层，以此类推。每个层都带有两个属性：前进指针和跨度。前进指针用于访问位于表尾方向的其他节点，而跨度则记录了前进指针所指向节点和当前节点的距离。在上图中，连线上带数字的箭头就代表前进指针，而那个数字就是跨度。当程序从表头向表尾进行遍历时，访问会沿着层的前进指针进行。
后退指针：节点中用BW字样标记节点的后腿指针，它指向位于当前节点的前一个节点，后退指针在程序从表尾项表头遍历时使用。
分值：各个节点中的1.0、2.0和3.0就是节点所保存的分值。在跳跃表中，节点按各自保存的分值从小到大排列。
成员对象：各个节点中的o1、o2和o3是节点所保存的成员对象。

注意表头节点和其他节点的构造是一样的：表头节点也有后退指针、分值和成员对象。只是不会用到这些属性而已。

注意这里展示的表头最大层数为32(Redis3.0)，这是根据ZSKIPLIST_MAXLEVEL决定的，在Redis5.x中这个值为64.

节点源码

跳跃表节点定义如下：

typedef struct zskiplistNode {
    //成员对象
    robj *obj;
    //分值
    double score;
    //后退指针
    struct zskiplistNode *backward;
    //层
    struct zskiplistLevel {
        //前进指针
        struct zskiplistNode *forward;
        //跨度
        unsigned int span;
    } level[];
} zskiplistNode;

层

/**
为我们将要创建的zskiplistNode返回一个随机level。这个函数的返回值在1和ZSKIPLIST_MAXLEVEL之间(包括这两个值)，使用类似于幂律的分布，较高的级别不太可能被返回。ZSKIPLIST_P 0.25 */
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

跨度

层的跨度用于记录两个节点之间的距离:

两个节点之间的跨度越大，它们相距越远
指向NULL的所有前进指针跨度都为0，因为它们没有连向任何节点

初步看上去决定跨度有什么用??

跨度实际上是用来计算排位的：在查找某个节点的过程中，将沿途访问过的所有层的跨度累计起来，得到的结果就是目标节点在跳跃表中的排位。

分值和成员

跳跃表源码

跳跃表定义如下：

typedef struct zskiplist {
    //表头和表尾节点
    struct zskiplistNode *header, *tail;
    //表中节点数量
    unsigned long length;
    //表中最大层数
    int level;
} zskiplist;

使用一个zskiplist结构来持有跳跃表节点，程序可以更方便地对整个跳跃表进行处理。

些许源码阅读

创建节点

//根据传递过来的level、score和obj创建一个节点
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
    //节点的大小=一个节点大小+level*一个层的大小
    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));
    //初始最大level为1
    zsl->level = 1;
    zsl->length = 0;
    //创建头节点，头节点层数为ZSKIPLIST_MAXLEVEL
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    //初始化头节点的所有level
    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;
}

插入节点

不难知道，跳跃表插入节点需要经过以下步骤：

查找要插入的位置
调整跳跃表高度
插入节点
调整后退指针

以插入下方跳跃表为例(level为2，length为3)，插入节点score为31，层高为3。
请输入图片描述

查找位置

查找待插入节点位置的源代码如下所示：

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;
    }
    update[i] = x;
}

重点关注update[ZSKIPLIST_MAXLEVEL]和rank[ZSKIPLIST_MAXLEVEL]这两个数组，特别重要。

update[]：插入节点时，需要更新被插入界定啊每层的前一个节点。由于每层更新的节点不一样，所以将每层需要更新的节点记录在update[i]中。
rank[]：记录当前层从header节点到update[i]节点所经历的跨度，在更新update[i]的跨度和设置新插入节点的跨度时用到。

查找节点的插入位置逻辑如下所示：

初始状态：x = head,i = zsl->level - 1 = 1;
第一次循环：
- rank[1] = 1 == 1 ? 0 : rank[i+1];=>rank[1] = 0;
- x->level[1].forward存在，并且x->level[1].forward.score小于31，说明待插入节点在其后面，进入while循环
- 更新rank，rank[1] += x->level[1].span;=>rank[1] = 1
- x向前进一个节点，成为第一个节点
- 第一个节点的前进指针(x->level[1].forward)为NULL，不进入while循环。update[1]为第一个节点
第二次循环：
- i=0，x为第一个节点
- i与zsl-level-1不相等，rank[0]=rank[0+1]=1
- x->level[0].forward存在，并且x->level[0].forward->score小于31，进入while循环
- 更新rank，rank[0] += x->level[0].span;=>rank[0] = 2
- x向前进一个节点，成为第二个节点
- x->level[0].forward存在，但是x->level[0].forward->score大于41，不再进入while循环。update[0]为第二个节点。

请输入图片描述

调整跳跃表高度

如果插入节点的高度(假设为3)大于跳跃表的高度，我们需要调整跳跃表的高度。

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);//创建带插入节点
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;
}

请输入图片描述

调整backward

x->backward = (update[0] == zsl->header) ? NULL : update[0];//x插入到第一个的话backward为NULL
if (x->level[0].forward)
    x->level[0].forward->backward = x;//下一个节点的backward指针指向自己
else
    zsl->tail = x;//x插入到最后了，需要更新tail
zsl->length++;
return x;

请输入图片描述

跳跃表的应用

在Redis中，跳跃表主要应用于有序集合的底层实现(有序集合的另一种实现方式为压缩链表)。
Redis的配置文件中关于有序集合底层实现的两个配置：

# 与哈希和列表类似，排序集也经过特殊编码，以节省大量空间。
# 这种编码只在排序集的长度和元素低于以下限制时使用:
zset-max-ziplist-entries 128 # 采用压缩列表时，元素个数最大值
zset-max-ziplist-value 64    # 采用压缩列表时，每个元素的字符串长度最大值

采用哪种底层实现在t_zset.c的zaddGenericCommand函数中说明了：

//zset-max-ziplist-entries的值是否为0
//zset-max-ziplist-value小于要插入元素的字符串长度
if (server.zset_max_ziplist_entries == 0 ||
`   server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
{
    zobj = createZsetObject(); //使用跳跃表实现
} else {
    zobj = createZsetZiplistObject(); //使用压缩列表实现
}

zset中元素个数大于zset-max-ziplist-entries
插入元素的字符串长度大于zset-max-ziplist-value

一旦满足以上任意一条件则会将压缩列表转为跳跃表：

if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
    zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
    zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);

值得注意的是，zset在转为跳跃表后，即使元素被逐渐删除，也不会重新转为压缩列表。

版权属于：带翅膀的猫
本文链接：https://www.chengpengper.cn/archives/111/
转载时须注明出处及本声明

跳跃表

Redis的跳跃表

节点源码

层

跨度

分值和成员

跳跃表源码

些许源码阅读

创建节点

创建跳跃表

插入节点

查找位置

调整跳跃表高度

插入节点

调整backward

跳跃表的应用