跳表(SkipList)原理与Redis ZSET应用实战

1. 真实场景：线上ZSET突然变慢

2024年3月，我们一个排行榜服务出现ZRANK耗时从50us飙升到3ms。业务数据量约100万，每个用户每月重算一次积分。排查发现：有些用户积分相同导致ZRANK需要遍历相同score的节点链表。原来Redis ZSET底层是跳表+字典，当大量相同score时，跳表退化为链表？我们深入源码后发现：跳表在score相同时按member字典序比较，插入复杂度稳定O(logN)，但ZRANK在score相等时需在skiplist中沿level[0]链表顺序查找，最坏O(N)。那个问题其实是业务侧插入大量重复score导致。但这触发了我对跳表源码的好奇。今天就用这篇文章彻底拆解跳表，并带你用Go手写一个可运行的跳表，再对比Redis源码实现。

2. 跳表原理

跳表（Skip List）由William Pugh于1990年提出，是一种可以替代平衡树的数据结构。它基于多层链表，上层链表是下层链表的“快速通道”。每个节点随机决定其层数，使得期望查找、插入、删除时间复杂度均为O(logN)。

核心结构：

• 层（Level）：每个节点包含一个level数组，每个level包含一个指向同层下一节点的forward指针（有的实现还有backward指针）。
• 头节点：固定层数MAX_LEVEL，指向每一层的第一个实际节点。
• 随机层数：常用概率p=1/2，每层概率为p。最大层数通常限制为32或64。

查找过程：从头节点最高层开始，若下一节点key小于目标，则向右移动；否则向下移动一层。重复直到最底层找到目标。

3. 方案对比：跳表 vs 红黑树 vs 有序数组

我们对比三种常见有序集合实现：

跳表核心优势在于：随机化使得代码简单；层数可控；并发下仅需修改指针。Redis选择跳表而非红黑树，主要就是因为跳表更易实现、调试和内存回收。

4. Go实现完整跳表（可运行）

我们用Go实现一个支持int key的跳表，包含插入、查找、删除、打印所有元素。代码基于Redis风格，但简化了member比较。完整代码在github链接，以下展示核心结构。

4.1 节点结构

// skiplist_node.go
const MAX_LEVEL = 32
const P = 0.25 // 概率，1/4

type Node struct {
    key     int
    value   interface{}
    forward []*Node // 每层的前向指针
}

func newNode(key int, value interface{}, level int) *Node {
    return &Node{
        key:     key,
        value:   value,
        forward: make([]*Node, level),
    }
}

4.2 随机层数函数

// random_level.go
import "math/rand"

func randomLevel() int {
    level := 1
    for rand.Float64() < P && level < MAX_LEVEL {
        level++
    }
    return level
}

4.3 跳表结构及插入

// skip_list.go
type SkipList struct {
    header *Node
    level  int // 当前最大层数
}

func NewSkipList() *SkipList {
    return &SkipList{
        header: newNode(0, nil, MAX_LEVEL),
        level:  1,
    }
}

func (sl *SkipList) Insert(key int, value interface{}) {
    update := make([]*Node, MAX_LEVEL)
    x := sl.header
    // 从最高层向下搜索，记录每层需要更新的节点
    for i := sl.level - 1; i >= 0; i-- {
        for x.forward[i] != nil && x.forward[i].key < key {
            x = x.forward[i]
        }
        update[i] = x
    }
    // 检查最底层下一个节点是否等于key，是则更新值
    x = x.forward[0]
    if x != nil && x.key == key {
        x.value = value
        return
    }
    // 随机层数
    level := randomLevel()
    if level > sl.level {
        for i := sl.level; i < level; i++ {
            update[i] = sl.header
        }
        sl.level = level
    }
    newNode := newNode(key, value, level)
    // 更新每层的前向指针
    for i := 0; i < level; i++ {
        newNode.forward[i] = update[i].forward[i]
        update[i].forward[i] = newNode
    }
}

4.4 查找和删除

// search_delete.go
func (sl *SkipList) Search(key int) (interface{}, bool) {
    x := sl.header
    for i := sl.level - 1; i >= 0; i-- {
        for x.forward[i] != nil && x.forward[i].key < key {
            x = x.forward[i]
        }
    }
    x = x.forward[0]
    if x != nil && x.key == key {
        return x.value, true
    }
    return nil, false
}

func (sl *SkipList) Delete(key int) bool {
    update := make([]*Node, MAX_LEVEL)
    x := sl.header
    for i := sl.level - 1; i >= 0; i-- {
        for x.forward[i] != nil && x.forward[i].key < key {
            x = x.forward[i]
        }
        update[i] = x
    }
    x = x.forward[0]
    if x == nil || x.key != key {
        return false
    }
    for i := 0; i < sl.level; i++ {
        if update[i].forward[i] != x {
            break
        }
        update[i].forward[i] = x.forward[i]
    }
    // 减小层数
    for sl.level > 1 && sl.header.forward[sl.level-1] == nil {
        sl.level--
    }
    return true
}

4.5 测试代码

// main.go
package main

import (
    "fmt"
    "time"
    "math/rand"
)

func main() {
    sl := NewSkipList()
    // 插入10万个随机键值
    n := 100000
    rand.Seed(time.Now().UnixNano())
    start := time.Now()
    for i := 0; i < n; i++ {
        key := rand.Intn(1000000)
        sl.Insert(key, i)
    }
    fmt.Printf("插入 %d 个元素耗时: %v\n", n, time.Since(start))
    // 查找
    start = time.Now()
    for i := 0; i < n; i++ {
        key := rand.Intn(1000000)
        _, _ = sl.Search(key)
    }
    fmt.Printf("查找 %d 次耗时: %v\n", n, time.Since(start))
    // 删除一半
    start = time.Now()
    for i := 0; i < n/2; i++ {
        key := rand.Intn(1000000)
        sl.Delete(key)
    }
    fmt.Printf("删除 %d 个元素耗时: %v\n", n/2, time.Since(start))
}

5. Redis ZSET源码分析

Redis 7.2.4中，ZSET底层使用zskiplist + dict。结构体定义在server.h中：

// server.h (Redis 7.2.4)
typedef struct zskiplistNode {
    sds ele;
    double score;
    struct zskiplistNode *backward;
    struct zskiplistLevel {
        struct zskiplistNode *forward;
        unsigned long span; // 跨越的节点数，用于ZRANK快速计算排名
    } level[];
} zskiplistNode;

typedef struct zskiplist {
    struct zskiplistNode *header, *tail;
    unsigned long length;
    int level;
} zskiplist;

关键点：

• 每个节点包含一个level数组，每个level有forward和span。span表示从当前节点到下一节点跨越的底层节点数，用于快速计算ZRANK（排名）。
• 插入时更新span。
• score相同时按ele字典序比较。
• dict用于O(1)根据member查找节点，跳表用于范围操作和排序。

6. 效果数据：压测对比

使用Go实现的跳表与标准库的排序切片（二分插入）做对比。测试环境：Linux x86_64, 8核, Go1.22, 数据量10万随机int。

跳表插入比红黑树快约20%？实际测试中红黑树旋转开销略大，但差距不大。跳表删除也比红黑树简单。但有序切片插入太慢，不适用动态集合。

7. 避坑指南

• 随机层数种子：必须使用高质量随机数，math/rand即可，但需要初始化种子，否则每次层数分布一样导致性能退化。建议用crypto/rand生成种子，但性能稍慢。
• 层数上限设置：MAX_LEVEL=32对于2^32个元素足够。如果层数太小，高层分布不均，期望复杂度变差。
• 内存对齐：Go中结构体字段顺序会影响内存对齐，尽量将指针放在一起。我们的Node中forward切片本身是引用，不影响。
• 并发读写：跳表可以实现无锁并发读，但写操作需要加锁或CAS。Redis单线程所以没这个问题。自己实现时，插入/删除需要加写锁，读可无锁（读时线程安全？需谨慎，因为可能存在并发修改导致读到中间状态）。建议使用sync.RWMutex。
• 更新操作：Redis中的ZADD当score或member变更时，会删除旧节点再插入新节点，而不是原地修改。因为跳表节点不可变（key决定位置）。如果实现update，必须重新计算层数吗？不，删除原key再插入新key。
• 相同score的排序：Redis既按score升序，又按member字典序。如果业务有大量相同score，ZRANK会退化为O(N)的遍历（沿底层链表找相同score段内的member）。务必避免score重复高。
• span失效：在并发删除时，如果更新了span但其他线程还在读取，会导致ZRANK错误。Redis无此问题。若自己实现并发跳表，需要原子操作或加锁。

8. 总结

跳表是简单且高效的有序数据结构，Redis ZSET用跳表实现了O(logN)的插入/查找/排名操作。通过Go实现完整跳表可以加深理解。如果你准备自制内存数据库，跳表是比红黑树更好的选择——代码量少、易调试、易并发化。切记处理好随机层数、锁和span更新。

最后，回到开头的问题：线上ZRANK变慢是因为大量相同score导致同一score段内遍历链表。解决方案：设计score时确保唯一性（例如加上时间戳），或限制相同score数量。如果数据量大，可改用ZRANKBYSCORE？不，还是用ZRANGEBYSCORE配合LIMIT。好在Redis 7.0后引入了ZRANK支持lexicographic，但本质一样。最彻底的办法：业务层控制score分布。