Redis `skiplist` 跳跃表机制：有序集合 ZSet 的高效实现

好的，各位观众，欢迎来到今天的Redis专场！今天我们要聊的，是Redis里一个非常重要的底层数据结构——跳跃表（skiplist）。各位可能要问了，跳跃表是啥玩意？听起来像是一种新的滑雪方式？

别急，其实跳跃表跟滑雪没啥关系，但它确实能让数据检索的速度“飞起来”。它在Redis中扮演着关键角色，特别是作为有序集合（ZSet）的底层实现之一。今天，我们就来扒一扒这个“跳跃高手”的底裤，看看它到底是怎么让Redis如此高效的。

一、为啥需要跳跃表？

在深入了解跳跃表之前，我们先来思考一个问题：有序集合用什么数据结构来实现最好？

• 数组？ 插入、删除元素太慢，需要移动大量元素，复杂度是O(n)。
• 链表？ 查找元素只能遍历，复杂度也是O(n)。即使是有序链表，也只能一个个地比较。
• 平衡树 (比如红黑树)? 查找、插入、删除的复杂度都是O(log n)，效率很高。Redis早期确实考虑过用平衡树实现ZSet，但最终选择了跳跃表，这是为什么呢？

原因如下：

1. 实现难度： 跳跃表比平衡树更容易实现和维护。平衡树的代码通常比较复杂，容易出错。
2. 内存占用： 跳跃表的内存占用通常比平衡树更小。特别是当数据量较小的时候，跳跃表的优势更明显。
3. 范围查询： 跳跃表在范围查询方面具有优势。比如，要查找分数在某个范围内的所有元素，跳跃表可以快速定位到起始位置，然后顺序遍历即可。而平衡树则需要进行中序遍历，效率相对较低。
4. 可控性： 跳跃表的性能可以通过调整参数（比如层数）来控制，从而适应不同的应用场景。

总之，跳跃表在实现难度、内存占用、范围查询和可控性等方面都具有优势，因此被Redis选中，成为了ZSet的得力助手。

二、跳跃表：链表的华丽变身

跳跃表本质上是一种有序链表，但它通过引入多层索引，实现了快速查找。你可以把它想象成一个有多条“高速通道”的链表。

我们先来看一个普通的有序链表：

Head -> 1 -> 3 -> 5 -> 7 -> 9 -> 11 -> 13 -> 15 -> Tail

如果我们要查找元素9，只能从Head开始，逐个比较，直到找到9为止。这效率太低了！

现在，我们给这个链表加上一级索引：

Head -> 1 -> 3 -> 5 -> 7 -> 9 -> 11 -> 13 -> 15 -> Tail
        ^     ^     ^     ^     ^
Level 1: 1     5     9    13

Level 1这一层，每隔两个元素提取一个作为索引。现在，我们要查找9，就可以先在Level 1上查找：

1. 从Level 1的Head开始，找到1。
2. 1 < 9，继续往后找，找到5。
3. 5 < 9，继续往后找，找到9。
4. 找到了！

可以看到，通过一级索引，我们跳过了一些不必要的比较，提高了查找效率。

如果数据量很大，一级索引可能还不够快。我们可以再加一级索引，也就是二级索引：

Head -> 1 -> 3 -> 5 -> 7 -> 9 -> 11 -> 13 -> 15 -> Tail
        ^     ^     ^     ^     ^
Level 1: 1     5     9    13
        ^           ^
Level 2: 1           9

现在，我们查找9，就可以先在Level 2上查找：

1. 从Level 2的Head开始，找到1。
2. 1 < 9，继续往后找，找到9。
3. 找到了！

通过二级索引，我们又跳过了一些比较。以此类推，我们可以构建更多层的索引，从而实现更快的查找速度。

这就是跳跃表的基本思想：通过多层索引，将查找的时间复杂度从O(n)降低到O(log n)。

三、Redis跳跃表的结构

Redis跳跃表的实现在细节上有一些优化。我们来看一下Redis跳跃表的结构定义（简化版）：

typedef struct zskiplistNode {
    sds ele; // 存储的字符串值
    double score; // 分值，用于排序
    struct zskiplistNode *backward; // 后退指针，方便倒序遍历
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 前进指针，指向下一个节点
        unsigned int span; // 跨度，表示当前节点和下一个节点之间的节点数
    } level[]; // 层级数组，每个节点可以有多层索引
} zskiplistNode;

typedef struct zskiplist {
    struct zskiplistNode *header; // 跳跃表的头节点
    struct zskiplistNode *tail; // 跳跃表的尾节点
    unsigned long length; // 跳跃表的长度，即节点数量
    int level; // 跳跃表的层数，即最高索引的层数
} zskiplist;

• zskiplistNode: 表示跳跃表中的一个节点。
  • ele: 存储的字符串值。在ZSet中，存储的是member。
  • score: 分值，用于排序。ZSet正是通过score来实现有序的。
  • backward: 后退指针，指向前一个节点。这个指针方便我们进行倒序遍历。
  • level[]: 层级数组。每个节点可以有多层索引。level[i].forward指向该节点在第i层的下一个节点，level[i].span表示该节点在第i层和下一个节点之间的节点数。
• zskiplist: 表示一个跳跃表。
  • header: 指向跳跃表的头节点。头节点不存储任何实际数据，它的作用是作为跳跃表的起始点。
  • tail: 指向跳跃表的尾节点。
  • length: 记录跳跃表的长度，即节点数量。
  • level: 记录跳跃表的层数，即最高索引的层数。

重要概念：跨度 (span)

跨度是跳跃表中一个很重要的概念。它表示当前节点和下一个节点之间的节点数。例如：

Level 2: H -> 9 -> Tail   (H是Header，头部节点)
              ^
      span=9  |

上图中，Header到9的跨度是9，表示从Header到9之间有9个节点（不包括Header，包括9）。

跨度的作用：

• 计算排名： 通过累加各层的跨度，可以快速计算出某个节点的排名。
• 优化查找： 在查找过程中，可以根据跨度来判断是否需要继续往后查找。

四、跳跃表的核心操作

跳跃表的核心操作包括：插入、删除和查找。

1. 插入 (Insert)

插入操作是跳跃表中最复杂的操作之一。它需要完成以下几个步骤：

1. 查找插入位置： 从最高层开始，逐层查找，找到每一层中，新节点应该插入的位置。
2. 确定新节点的层数： 新节点的层数不是固定的，而是随机生成的。通常使用抛硬币的方式来确定层数，例如，每次抛硬币，如果是正面，则层数加1，直到达到最大层数或者抛到反面为止。
3. 插入节点： 将新节点插入到每一层的合适位置。
4. 更新跨度： 更新每一层中，相关节点的跨度。

我们来看一段伪代码：

def insert(skiplist, score, ele):
    # 1. 查找插入位置
    update = [None] * skiplist.level  # 记录每一层需要更新的节点
    x = skiplist.header
    for i in range(skiplist.level - 1, -1, -1):
        while x.level[i].forward and (x.level[i].forward.score < score or 
                                       (x.level[i].forward.score == score and x.level[i].forward.ele < ele)):
            x = x.level[i].forward
        update[i] = x

    # 2. 确定新节点的层数
    level = randomLevel()  # 随机生成一个层数

    # 3. 如果新节点的层数大于当前跳跃表的层数，则更新跳跃表的层数
    if level > skiplist.level:
        for i in range(skiplist.level, level):
            update[i] = skiplist.header
            skiplist.header.level[i].span = skiplist.length
        skiplist.level = level

    # 4. 创建新节点
    x = createNode(level, score, ele)

    # 5. 插入节点
    for i in range(level):
        x.level[i].forward = update[i].level[i].forward
        update[i].level[i].forward = x

        # 6. 更新跨度
        x.level[i].span = update[i].level[i].span - (skiplist.length - getRank(skiplist, score, ele, i, update))
        update[i].level[i].span = (skiplist.length - getRank(skiplist, score, ele, i, update)) + 1

    # 7. 更新后退指针
    if update[0] == skiplist.header:
        x.backward = None
    else:
        x.backward = update[0]
    if x.level[0].forward:
        x.level[0].forward.backward = x
    else:
        skiplist.tail = x

    # 8. 更新跳跃表的长度
    skiplist.length += 1

2. 删除 (Delete)

删除操作与插入操作类似，也需要先查找要删除的节点，然后将其从每一层中移除，并更新跨度。

def delete(skiplist, score, ele):
    # 1. 查找要删除的节点
    update = [None] * skiplist.level
    x = skiplist.header
    for i in range(skiplist.level - 1, -1, -1):
        while x.level[i].forward and (x.level[i].forward.score < score or 
                                       (x.level[i].forward.score == score and x.level[i].forward.ele < ele)):
            x = x.level[i].forward
        update[i] = x

    # 2. 确认找到了要删除的节点
    x = x.level[0].forward
    if x and x.score == score and x.ele == ele:
        # 3. 从每一层中移除节点
        for i in range(skiplist.level):
            if update[i].level[i].forward == x:
                update[i].level[i].forward = x.level[i].forward
                update[i].level[i].span += x.level[i].span - 1
            else:
                update[i].level[i].span -= 1

        # 4. 更新后退指针
        if x.level[0].forward:
            x.level[0].forward.backward = x.backward
        else:
            skiplist.tail = x.backward
        if x.backward:
            x.backward.level[0].forward = x.level[0].forward

        # 5. 更新跳跃表的层数
        while skiplist.level > 1 and skiplist.header.level[skiplist.level - 1].forward is None:
            skiplist.level -= 1

        # 6. 更新跳跃表的长度
        skiplist.length -= 1

3. 查找 (Search)

查找操作相对简单。从最高层开始，逐层查找，直到找到目标节点或者确定目标节点不存在。

def search(skiplist, score, ele):
    x = skiplist.header
    for i in range(skiplist.level - 1, -1, -1):
        while x.level[i].forward and (x.level[i].forward.score < score or 
                                       (x.level[i].forward.score == score and x.level[i].forward.ele < ele)):
            x = x.level[i].forward

    x = x.level[0].forward
    if x and x.score == score and x.ele == ele:
        return x
    else:
        return None

五、Redis跳跃表的优化

Redis跳跃表在实现上进行了一些优化，以提高性能和减少内存占用。

• 压缩列表 (ziplist) 优化： 当ZSet的元素数量较少，并且元素的大小也比较小的时候，Redis会使用压缩列表来代替跳跃表。压缩列表是一种紧凑的数据结构，可以有效地减少内存占用。
• 随机层数： 新节点的层数不是固定的，而是随机生成的。这样可以避免跳跃表退化成链表。Redis使用一个概率值P来控制层数的生成。P的值越小，跳跃表的层数越高，查找效率越高，但内存占用也越大。Redis默认的P值为0.25。
• 后退指针： 跳跃表中的每个节点都有一个后退指针，指向前一个节点。这个指针方便我们进行倒序遍历。

六、跳跃表的应用场景

跳跃表在Redis中主要用于实现有序集合（ZSet）。ZSet是一种常用的数据结构，它可以存储一组有序的元素，并支持快速的查找、插入和删除操作。

ZSet的应用场景非常广泛，例如：

• 排行榜： 可以使用ZSet来存储用户的积分，并根据积分进行排序，从而实现排行榜功能。
• 计数器： 可以使用ZSet来存储计数器的值，并根据计数器的值进行排序。
• 带权重的队列： 可以使用ZSet来实现带权重的队列，每个元素都有一个权重，权重越高，优先级越高。
• 范围查询： ZSet支持范围查询，可以快速查找分数在某个范围内的所有元素。

七、跳跃表 vs. B树/B+树

跳跃表和B树/B+树都是常用的索引结构，它们各有优缺点。

特性	跳跃表	B树/B+树
实现难度	相对简单	相对复杂
内存占用	通常比B树/B+树小（特别是数据量较小的时候）	通常比跳跃表大
范围查询	优势明显，可以快速定位到起始位置，然后顺序遍历	需要进行中序遍历，效率相对较低
写入性能	相对较好，因为插入和删除操作不需要像B树/B+树那样进行复杂的树结构调整	可能需要进行树结构调整，写入性能相对较差
并发支持	相对容易实现并发，因为节点之间的锁竞争较少	并发控制相对复杂，需要考虑锁的粒度和死锁问题
磁盘IO	主要用于内存数据库，磁盘IO较少	主要用于磁盘数据库，需要频繁进行磁盘IO
适用场景	内存数据库，需要快速查找和范围查询，对写入性能要求较高，并发量较小	磁盘数据库，需要存储大量数据，对查询性能要求较高，并发量较大
空间复杂度	O(n), n是元素的数量。但实际使用中，由于概率因素，平均空间复杂度会略大于n	B树和B+树的空间复杂度主要取决于树的阶数和数据的填充率，通常为O(n)。
时间复杂度	查找、插入、删除的平均时间复杂度为O(log n)。	查找、插入、删除的时间复杂度为O(log n)。

八、总结

跳跃表是一种高效的数据结构，它通过引入多层索引，实现了快速查找。它在Redis中扮演着关键角色，特别是作为有序集合（ZSet）的底层实现之一。

跳跃表具有以下优点：

• 实现简单
• 内存占用小
• 范围查询效率高
• 可控性强

当然，跳跃表也有一些缺点：

• 空间复杂度略高

总的来说，跳跃表是一种非常优秀的数据结构，它在Redis中得到了广泛的应用。

好了，今天的Redis跳跃表之旅就到这里了。希望大家通过今天的学习，能够对跳跃表有更深入的了解。下次再见！