从redis源码看数据结构（四）跳跃链表

本是古典何须时尚 2023-06-14 13:58 153阅读 0赞

> 笔者大三，最近复习到了redis，如有错误，还请及时指出

# 从redis源码看数据结构（四）跳跃链表 #

### 文章目录 ###

*  从redis源码看数据结构（四）跳跃链表
 *   *  一，redis中的跳表
     *   *  1.底层结构体
     *  二，redis中跳跃链表的操作
     *   *  1.创建跳表
         *  2.插入一个节点
         *  3.删除节点
     *  三，redis中的跳表和普通跳表的区别

## 一，redis中的跳表 ##

redis 中的有序集合是由我们之前介绍过的字典加上跳表实现的，字典中保存的数据和分数 score 的映射关系，每次插入数据会从字典中查询key，如果已经存在了，就不再插入，有序集合中是不允许重复数据。

typedef struct zset {
        
        dict *dict;
        zskiplist *zsl;
    } zset;

![在这里插入图片描述][20191119212312646.png]

### 1.底层结构体 ###

typedef struct zskiplist {
        
    
        // 表头节点和表尾节点(指向的是最底层链表的头结点和尾节点)
        struct zskiplistNode *header, *tail;
    
        // 表中节点的数量
        unsigned long length;
    
        // 表中层数最大的节点的层数
        int level;
    
    } zskiplist;

跳表节点

typedef struct zskiplistNode {
        
    
        // 后退指针
        struct zskiplistNode *backward;
    
        // 分值，最底层链表是按照分支大小，进行排序串联起来的
        double score;
    
        // 成员对象
        //这里的redis版本是4.0，以前是redisObject类型，现在是sds类型，即现在跳表只用于存储字符串数据
       sds ele;
    
        //每个节点除了储存节点自身数据外，还通过level数组保存了该节点在整个跳表各个索引层的节点引用
        // 层
        struct zskiplistLevel {
        
    
            // 前进指针
            struct zskiplistNode *forward;
    
            // 跨度，跨过多少个节点
            unsigned int span;
    
        } level[];
    
    } zskiplistNode;

关于zskiplistLevel的具体结构是这样的：

![image][]

整张表的基本结构是这样的：

![image][image 1]

## 二，redis中跳跃链表的操作 ##

### 1.创建跳表 ###

/*
     * 创建一个跳跃表
     *
     * T = O(1)
     */
    zskiplist *zslCreate(void) {
        
        int j;
        zskiplist *zsl;
    
        zsl = zmalloc(sizeof(*zsl));
    	//默认一层索引层
        zsl->level = 1;
        //初始化时没有节点
        zsl->length = 0;
    
        // 初始化头节点， O(1)，给其分配32个索引层，即redis中索引层最多32层ZSKIPLIST_MAXLEVEL = 32
        zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
        // 初始化层指针，O(1)
        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;
    }

redis 中实现的跳表最高允许 32 层索引，这么做也是一种性能与内存之间的衡量，过多的索引层必然占用更多的内存空间，32 是一个比较合适值。

### 2.插入一个节点 ###

/*
     * 将包含给定 score 的对象 obj 添加到 skiplist 里
     *
     * T_worst = O(N), T_average = O(log N)
     */
    zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
        
    
        //update数组将用于记录新节点在每一层索引的目标插入位置
        zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    
        // 记录寻找元素过程中，每层所跨越的节点数
        unsigned int rank[ZSKIPLIST_MAXLEVEL];
    
        int i, level;
    
        redisAssert(!isnan(score));
        x = zsl->header;
        //这一段就是遍历每一层索引，找到最后一个小于当前给定score值的节点，保存在update数组中 
        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 &&                   
                // 右节点的 score 比给定 score 小
                (x->level[i].forward->score < score ||      
                    // 右节点的 score 相同，但节点的 member 比输入 member 要小
                    (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数组中已经记录好，每一层最后一个小于给定score值的节点
        
        // 计算新的随机层数
        level = zslRandomLevel();
        // 如果 level 比当前 skiplist 的最大层数还要大
        //为高出来的索引层赋初始值，update[i]指向哨兵节点，想构造跳跃链表一样
        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);
        // 根据 update 和 rank 两个数组的资料，初始化新节点
        // 并设置相应的指针
        // O(N)
        for (i = 0; i < level; i++) {
        
            //原： update[i]->level[i] -> update[i]->level[i].forward
            x->level[i].forward = update[i]->level[i].forward;
            update[i]->level[i].forward = x;
            //后： update[i]->level[i] -> x -> update[i]->level[i].forward
    
            //rank[0]等于新节点再最底层链表的排名，就是它前面有多少个节点
            //update[i]->level[i].span记录的是目标节点与后一个索引节点之间的跨度，即跨越了多少个节点
            //得到新插入节点与后一个索引节点之间的跨度
            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 */
        // 更新沿途访问节点的 span 值
        for (i = level; i < zsl->level; i++) {
        
            update[i]->level[i].span++;
        }
    
        // 设置后退指针
        x->backward = (update[0] == zsl->header) ? NULL : update[0];
        // 设置 x 的前进指针
        if (x->level[0].forward)
            x->level[0].forward->backward = x;
        else
            // 这个是新的表尾节点
            zsl->tail = x;
    
        // 更新跳跃表节点数量
        zsl->length++;
    
        return x;
    }

大概逻辑：

1.  从最高索引层开始遍历，根据 score 找到它的前驱节点，用 update 数组进行保存
2.  每一层得进行节点的插入，并计算更新 span 值
3.  修改 backward 指针与 tail 指针

### 3.删除节点 ###

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
    /*
     * 节点删除函数
     *
     * T = O(N)
     */
    void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
        
        int i;
    
        // 修改相应的指针和 span , O(N)
        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;
            }
        }
    
        // 处理表头和表尾节点
        if (x->level[0].forward) {
        
            x->level[0].forward->backward = x->backward;
        } else {
        
            zsl->tail = x->backward;
        }
    
        // 收缩 level 的值, O(N)
        while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
            zsl->level--;
    
        zsl->length--;
    }

## 三，redis中的跳表和普通跳表的区别 ##

redis的跳表和普通的跳表实现没有多大区别，主要区别在三处：

*  redis的跳表引入了score，且score可以重复
 *  排序不止根据分数，还可能根据成员对象（当分数相同时）
 *  有一个前继指针，因此在第1层，就形成了一个双向链表，从而可以方便的从表尾向表头遍历

[20191119212312646.png]: https://img-blog.csdnimg.cn/20191119212312646.png
[image]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91c2VyLWdvbGQtY2RuLnhpdHUuaW8vMjAxOS8xMC8xMi8xNmRiZmZlNTllYTg0ZmI5?x-oss-process=image/format,png
[image 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91c2VyLWdvbGQtY2RuLnhpdHUuaW8vMjAxOS8xMC8xMi8xNmRiZmZlNWIzZTcwYjRk?x-oss-process=image/format,png