Redis源码解析：数据结构详解-skiplist

跳表是个什么数据结构？

本文的很多内容参考自如下文章《Redis 为什么用跳表而不用平衡树？》，为了加深理解，所以用自己的话复述一遍。

如图所示，redis中的zset在元素少的时候用ziplist来实现，元素多的时候用skiplist和dict来实现。那么skiplist是一个什么样的数据结构呢？

skiplist（跳表）是一种为了加速查找而设计的一种数据结构。它是在有序链表的基础上发展起来的。

如下图是一个有序链表（最左侧的灰色节点为一个空的头节点），当我们要插入某个元素的时候，需要从头开始遍历直到找到该元素，或者找到第一个比给定元素大的数据（没找到），时间复杂度为O(n)

假如我们每隔一个节点，增加一个新指针，指向下下个节点，如下图所示。这样新增加的指针又组成了一个新的链表，但是节点数只有原来的一半。

当我们想查找某个数据的时候可以现在新链表上进行查找，当碰到比要查找的数据大的节点时，再到原来的链表上进行查找。

如下为查找23的过程，查找的过程为图中红色箭头指向的方向

1. 23首先和7比较，然后和19比较，都比他们大，接着往下比。23和26比较，比26小。然后从19这节点回到原来的链表
2. 和22比较，比22大，继续和下一个节点26比，比26小，说明23在跳表中不存在

利用上面的思路，我们可以在新链表的基础每隔一个节点再生成一个新的链表。

在新链表的基础上，还是查找23，先和19比较，比19大，并且19的下一个节点为NULL，从，从19条跳到下一层节点去查找，可以看到当链表比较长时，这种方式可以让我们跳过很多下层节点，大大加快查找的速度。

skiplist就是按照这种思想设计出来的，当然你可以每隔3，4等向上抽一层节点。但是这样做会有一个问题，如果你严格保持上下两层的节点数为1：2，那么当新增一个节点，后续的节点都要进行调整，会让时间复杂度退化到O（n），删除数据也有同样的问题。

skiplist为了避免这种问题的产生，并不要求上下两层的链表个数有着严格的对应关系，而用随机函数得到每个节点的层数。比如一个节点随机出的层数为3，那么把他插入到第一层到第三层这3层链表中。为了方便理解，下图演示了一个skiplist的生成过程

由于层数是每次随机出来的，所以新插入一个节点并不会影响其他节点的层数。插入一个节点只需要修改节点前后的指针即可，降低了插入的复杂度。

刚刚创建的skiplist包含4层链表，假设我们依然查找23，查找路径如下。插入的过程也需要经历一个类似查找的过程，确定位置后，再进行插入操作

和随机函数相关的2个常量值如下

// 大概每隔多少个节点向上提取一层的比例
// 1/2为大概每隔2个节点向上提取一层，1/4为每隔4个节点向上提取一层
private static final float SKIPLIST_P = 0.5f;
// 最高层数为16
private static final int MAX_LEVEL = 16;

计算随机层数的代码如下

// 理论来讲，一级索引中元素个数应该占原始数据的 50%，二级索引中元素个数占 25%，三级索引12.5% ，一直到最顶层。
// 因为这里每一层的晋升概率是 50%。对于每一个新插入的节点，都需要调用 randomLevel 生成一个合理的层数。
// 该 randomLevel 方法会随机生成 1~MAX_LEVEL 之间的数，且 ：
// 50%的概率返回 1
// 25%的概率返回 2
// 12.5%的概率返回 3 ...
// Math.random()返回的值为[0, 1)
private int randomLevel() {
    int level = 1;

    while (Math.random() < SKIPLIST_P && level < MAX_LEVEL)
        level += 1;
    return level;
}

一个完整的跳表实现如下（代码地址：https://github.com/wangzheng0822/algo/tree/master/java/17_skiplist）

public class SkipList {

    // 大概每隔多少个节点向上提取一层的比例
    // 1/2为大概每隔2个节点向上提取一层，1/4为每隔4个节点向上提取一层
    private static final float SKIPLIST_P = 0.5f;
    // 最高层数为16
    private static final int MAX_LEVEL = 16;

    // 目前的层数
    private int levelCount = 1;

    private Node head = new Node();  // 带头链表

    public Node find(int value) {
        Node p = head;
        // p.forwards[i]表示节点p到第i层的下一个节点
        for (int i = levelCount - 1; i >= 0; --i) {
            while (p.forwards[i] != null && p.forwards[i].data < value) {
                p = p.forwards[i];
            }
        }
        // 遍历到最底层了
        if (p.forwards[0] != null && p.forwards[0].data == value) {
            return p.forwards[0];
        } else {
            return null;
        }
    }

    public void insert(int value) {
        int level = randomLevel();
        Node newNode = new Node();
        newNode.data = value;
        newNode.maxLevel = level;
        Node update[] = new Node[level];
        for (int i = 0; i < level; ++i) {
            update[i] = head;
        }

        // record every level largest value which smaller than insert value in update[]
        Node p = head;
        for (int i = level - 1; i >= 0; --i) {
            while (p.forwards[i] != null && p.forwards[i].data < value) {
                p = p.forwards[i];
            }
            update[i] = p;// use update save node in search path
        }

        // in search path node next node become new node forwords(next)
        for (int i = 0; i < level; ++i) {
            newNode.forwards[i] = update[i].forwards[i];
            update[i].forwards[i] = newNode;
        }

        // update node hight
        // 更新目前的层数
        if (levelCount < level) levelCount = level;
    }

    public void delete(int value) {
        Node[] update = new Node[levelCount];
        Node p = head;
        for (int i = levelCount - 1; i >= 0; --i) {
            while (p.forwards[i] != null && p.forwards[i].data < value) {
                p = p.forwards[i];
            }
            update[i] = p;
        }

        if (p.forwards[0] != null && p.forwards[0].data == value) {
            for (int i = levelCount - 1; i >= 0; --i) {
                if (update[i].forwards[i] != null && update[i].forwards[i].data == value) {
                    update[i].forwards[i] = update[i].forwards[i].forwards[i];
                }
            }
        }

        while (levelCount > 1 && head.forwards[levelCount] == null) {
            levelCount--;
        }

    }

    // 理论来讲，一级索引中元素个数应该占原始数据的 50%，二级索引中元素个数占 25%，三级索引12.5% ，一直到最顶层。
    // 因为这里每一层的晋升概率是 50%。对于每一个新插入的节点，都需要调用 randomLevel 生成一个合理的层数。
    // 该 randomLevel 方法会随机生成 1~MAX_LEVEL 之间的数，且 ：
    // 50%的概率返回 1
    // 25%的概率返回 2
    // 12.5%的概率返回 3 ...
    // Math.random()返回的值为[0, 1)
    private int randomLevel() {
        int level = 1;

        while (Math.random() < SKIPLIST_P && level < MAX_LEVEL)
            level += 1;
        return level;
    }

    public void printAll() {
        Node p = head;
        while (p.forwards[0] != null) {
            System.out.print(p.forwards[0] + " ");
            p = p.forwards[0];
        }
        System.out.println();
    }

    public class Node {
        private int data = -1;
        private Node forwards[] = new Node[MAX_LEVEL];
        private int maxLevel = 0;

        @Override
        public String toString() {
            StringBuilder builder = new StringBuilder();
            builder.append("{ data: ");
            builder.append(data);
            builder.append("; levels: ");
            builder.append(maxLevel);
            builder.append(" }");

            return builder.toString();
        }
    }

}

zset命令实现分析

// 将一个或多个 member 元素及其 score 值加入到有序集 key 当中
ZADD key score member [[score member] [score member] ...]

// 移除有序集 key 中的一个或多个成员，不存在的成员将被忽略。
ZREM key member [member ...]

// 返回有序集 key 中成员 member 的排名。其中有序集成员按 score 值递增(从小到大)顺序排列。
// 排名以 0 为底，也就是说， score 值最小的成员排名为 0 。
ZRANK key member

// 返回有序集 key 中，成员 member 的 score 值。
ZSCORE key member

// 返回有序集 key 中，指定区间内的成员。
// 其中成员的位置按 score 值递增(从小到大)来排序。
// 具有相同 score 值的成员按字典序来排列。
ZRANGE key start stop [WITHSCORES]

// 返回有序集 key 中，指定区间内的成员。
// 其中成员的位置按 score 值递减(从大到小)来排列。
// 具有相同 score 值的成员按字典序的逆序排列。
ZREVRANGE key start stop [WITHSCORES]

// 返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员
// 有序集成员按 score 值递增(从小到大)次序排列。
ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]

// 返回有序集 key 中， score 值介于 max 和 min 之间(默认包括等于 max 或 min )的所有的成员。
// 有序集成员按 score 值递减(从大到小)的次序排列。
// 具有相同 score 值的成员按字典序的逆序(reverse lexicographical order )排列。
ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]

用一个例子来演示一下，用sorted set来存储代数课（algebra）的成绩表

127.0.0.1:6379> zrem algebra Alice Bob David
(integer) 3
127.0.0.1:6379> zadd algebra 87.5 Alice 
(integer) 1
127.0.0.1:6379> zadd algebra 65.5 Charles 
(integer) 1
127.0.0.1:6379> zadd algebra 78.0 David 
(integer) 1
127.0.0.1:6379> zadd algebra 93.5 Emily 
(integer) 1
127.0.0.1:6379> zadd algebra 87.5 Fred 
(integer) 1
127.0.0.1:6379> zrank algebra Alice
(integer) 2
127.0.0.1:6379> zscore algebra Alice
"87.5"
127.0.0.1:6379> zrange algebra 0 3 withscores
1) "Charles"
2) "65.5"
3) "David"
4) "78"
5) "Alice"
6) "87.5"
7) "Fred"
8) "87.5"
127.0.0.1:6379> zrangebyscore algebra 80.0 90.0 withscores
1) "Alice"
2) "87.5"
3) "Fred"
4) "87.5"

我们分析一下上面所设计到的几个命令

ZSCORE：数据对应的分数，skiplist不支持
ZRANK：根据数据查询它对应的排名，skiplist不支持
ZRANGE：按照返回集合某个区间的数据，skiplist不支持
ZRANGEBYSCORE：基于分数区间查询数据集合，skiplist支持，典型的范围查找

redis中zset的数据结构如下

// server.h
typedef struct zset {
    dict *dict;
    zskiplist *zsl;
} zset;

当数据量较少的时候，sorted set用ziplist来实现
当数据量较多的时候，sorted set由一个叫做zset的数据结构来实现，这个zset包含一个dict和一个zskiplist，dict保存了数据到分数（score）的对应关系，skiplist用来根据分数查询数据

我们接着来分析一下zset是怎么实现上面的功能的

ZSCORE：数据对应的分数，skiplist不支持，用dict来实现
ZRANK：先在dict中由数据查到分数，再拿分数去扩展后的skiplist中去查找，查到的同时就获得了排名
ZRANGE：按照排名返回集合某个区间的数据，由扩展后的skiplist支持
ZRANGEBYSCORE：基于分数区间查询数据集合，skiplist支持，典型的范围查找

Redis中skiplist和经典的skiplist相比，有如下的不同

1. 分数允许重复，在经典的skiplist中是不允许的。当多个元素的分数相同时，还需要根据内容按照字典序排序
2. 第一层链表不是一个单向链表，而是一个双向链表，方便以倒序的方式获取一个范围内的元素
3. 可以很方便的计算出每个元素的排名

skiplist数据结构定义

zskiplistNode定义了skiplist的节点结构

// server.h
// 最高层数为64层
#define ZSKIPLIST_MAXLEVEL 64 /* Should be enough for 2^64 elements */
// 大概每隔4个节点向上抽象一层
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */

typedef struct zskiplistNode {
	// 字符串类型的member值
    sds ele;
    // 分值
    double score;
    // 后向指针
    struct zskiplistNode *backward;
    struct zskiplistLevel {
    	// 前向指针
        struct zskiplistNode *forward;
        // 跨度
        unsigned long span;
    } level[];
} zskiplistNode;

最上面的两个常量，ZSKIPLIST_MAXLEVEL和ZSKIPLIST_P主要用来控制构建策略。

ele：保存字符串类型的member值
score：保存分值
backward：后向指针，指向链表的上一个节点，在zskiplist中，只有最底层的一层链表为双向链表
level：前向指针数组，存放各层链表的下一个节点的指针。span保存了当前指针跨越了多少个节点，用来计算排名

zskiplist是真正的跳表结构

typedef struct zskiplist {
	// 跳表头尾指针
    struct zskiplistNode *header, *tail;
    // 节点的数量
    unsigned long length;
    // 跳表目前最高层数
    int level;
} zskiplist;

header：跳表头节点，是一个特殊节点，level数组元素个数为64，头节点在有序集合中存储member值和score值，ele为null，socre为0，不计入跳表总长度。头节点在初始化时64个元素的forward都指向null，span值为0
tail：跳表尾节点
length：跳表总长度，除头节点之外的节点总数
level：跳表目前最高层数，除去头节点

以上面的代数成绩表为例。Redis中一种可能的skiplist结构如下图

图中前向指针上的数字，表示对应的span值，即当前指针跨越了多少个节点。

当我们在这个skiplist中查找socre=89.0的元素（即Bob的成绩数据），查找过程中我们会跨越图中标红的指针，这些指针上面的span值累加起来，就得到了Bob的排名2+2+1-1=4（减1是因为rank从0开始）。当我们需要从大到小的排名，只需要用skiplist的长度减去查找路径上span的累加值，即6-(2+2+1)=1

skiplist中和排名相关的操作（获取某个值的排名，获取某个排名区间内的数据），都是通过累加span值来得到的。

Redis为什么用skipList来实现有序集合，而不是红黑树？

redis中zset常用的操作有如下几种，插入数据，删除数据， 查找数据，按照区间输出数据，输出有序序列。

1. 插入数据，删除数据，查找数据，输出有序序列这几种操作红黑树也能实现，时间复杂度和跳表一样。但是按照区间输出数据，红黑树的效率没有跳表高，跳表可以在O(logn)的时间复杂度定位区间的起点，然后向后遍历极客
2. 跳表和红黑树相比，比较容易理解，实现比较简单，不容易出错
3. 可以通过设置参数，改变索引构建策略，按需平衡执行效率和内存消耗

zset

先放member再放socre，按照score从小到大的顺序排列

参考博客

[1]https://juejin.cn/post/6844903446475177998
redis命令
[2]http://doc.redisfans.com/
大佬博客
[3]http://zhangtielei.com/posts/blog-redis-quicklist.html