本文中，我们围绕一个非常重要的Redis的内部数据结构 -- skiplist（跳跃表）展开讨论。Redis中使用跳跃表（skiplist）是为了实现sorted set（有序集合）这种对外的数据结构。sorted set提供的操作非常丰富，可以满足非常多的应用场景。搞清楚其底层实现是非常重要的，有利于我们在复杂场景中的使用。

skiplist简介

skiplist（跳跃表）是一种随机化的数据结构，是由William Pugh在论文《Skip lists：a probabilistic alternative to balanced trees》中提出，是一种层次化链表结构：查找、删除、添加等操作的时间复杂度O(logn)。

• 本质是解决查找问题

skiplist本质上是一种查询结构，用于解决算法中的查找问题（Searching），即：根据给定的key，快速查找到key所在的位置或者对应的value值。

先看一个普通的有序链表，其中最左侧的灰色结点表示一个空的头结点：

在这样一个有序链表中，由于链表是没法进行位置定位的，所以如果我们要查找某个数据，需要从头开始逐个进行比较，直至找到包含数据的那个结点或者第一个比给定数据大的结点（没有要找的数据），时间复杂度O(n)。同样的，当我们要插入新数据时，也需要同样的查找过程，从而确定插入位置，时间复杂度同样是O(n)。

如果每相邻的两个结点增加一个指针，让指针指向下一个结点，结果会如何呢？

这样新增的指针连城了一个新的链表，包含的结点个数只有原来的一半。现在，我们查找数据时，可以沿着新的链表进行查找，当遇到比待查数据大的结点时，再回到原来的链表中进行查找。例如，我们需要查找数据23，查找的路径是沿着下图中红色标记的指针方向进行的：

1. 首先和7进行比较，比7大，沿当前链表继续向后比较；
2. 沿着当前链表比较，26比23大，回到原链表，与22进行比较；
3. 22比23小，又比下一个比较数据26小，说明待查数据23在原链表中不存在，而且它的待插入位置在22和26之间。

由此，利用同样的方式，我们可以在上层在产生新的链表，继续为每相邻的两个结点增加指针，产生第三层链表，如下图：

基于这种多层链表设计，skiplist应运而生。但是上述产生链表的方式存在一个问题：上面每一层链表的结点个数是下一层结点个数的一半，查找过程类似二分查找，查找时间复杂度可以降到O(logn)。但是插入数据时，会打乱原先上下层结点个数严格的1:2的对应关系。如果要维持这种对应关系，需要将新插入的结点以及后面所有的结点进行调整，这会使得时间复杂度退化到O(n)，删除存在同样的问题。有没有什么办法解决呢？skiplist是如何设计的呢？

skiplist不要求上下相邻两层链表之间的结点个数有严格的对应关系，而是为每个结点随机生成一个层数（level）。比如，一个结点随机生成的层数是5，那么就将它链入到第1层到第5层链表中。为了更好的理解，下图展示一个skiplist生成的过程：

从上面skiplist的创建和插入过程可以看出，每一个节点的层数（level）是随机出来的，而且新插入一个节点不会影响其它节点的层数。因此，插入操作只需要修改插入节点前后的指针，而不需要对很多节点都进行调整。这就降低了插入操作的复杂度。

• 为什么要使用跳跃表

首先，由于sorted set要支持随机的插入和删除，所以它不适合使用数组这种数据结构来实现。另外，关于排序问题，我们很容易想到红黑树/平衡树这样的树型结构，为什么Redis不使用这样的数据结构呢？

1. 性能考虑：树型结构需要进行一些类似于rebalance操作，这样的操作可能涉及到整棵树的结构调整，相对的skiplist的变化只涉及局部；
2. 实现考虑：在复杂度相同的情况下，skiplist实现更简单；

Redis中的skiplist实现

了解了skiplist数据结构的设计过程，我们来看一下Redis中skiplist是如何实现的？

Redis中skiplist是由 zskiplistNode 和 zskiplist 两个结构定义。源码如下：

#define ZSKIPLIST_MAXLEVEL 32
#define ZSKIPLIST_P 0.25

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

typedef struct zskiplist {
    // 跳跃表头指针
    struct zskiplistNode *header, *tail;
  
    // 表中结点的数量
    unsigned long length;
  
    // 表中层数最大的结点的层数
    int level;
  
} zskiplist;

• 随机层数

对于新插入的结点，都需要调用一个随机算法分配一个合理的层数，源码zslRandomLevel中定义：

int zslRandomLevel(void) {
                   
    int level = 1;
                   
    while ((random() & 0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) {
        level += 1;
    }
    
    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

从上述源码中，可以直观看出：期望目标是50%的概率分配到level1，25%的概率分配到level2，以此类推，有1/2^63概率分配到最顶层，因为每一层的晋升概率都是50%。

【注意】：Redis中skiplist默认允许的最大层数是32.

• skiplist创建

skiplist创建的过程比较简单，源码zslCreate中定义：

zskiplist *zslCreate(void) {

    int j; 
    zskiplist *zsl; 
                     
    // 申请内存空间
    zsl = zmalloc(sizeof(*zsl));

    // 初始化层数为 1
    zsl->level = 1;
    
    // 初始化长度为 0 
    zsl->length = 0; 

    // 创建一个层数为 32，分数为 0，没有 value 值的跳跃表头节点 
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL); 

    // 跳跃表头节点初始化 
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) { 
      
          // 将跳跃表头节点的所有前进指针 forward 设置为 NULL 
          zsl->header->level[j].forward = NULL; 
      
          // 将跳跃表头节点的所有跨度 span 设置为 0 
          zsl->header->level[j].span = 0;
    } 

    // 跳跃表头节点的后退指针 backward 置为 NULL
    zsl->header->backward = NULL;

    // 表头指向跳跃表尾节点的指针置为 NULL 
    zsl->tail = NULL; 
    
    return zsl; 
}

• 插入操作实现
• 找到当前待插入的位置
• 创建新结点，调整前后指针指向，完成插入

Redis源码实现具体过程：

1. 声明存储变量

// 存储搜索路径
zskiplistNode *update[ZSKIPLIST_MAXLEVEL],
*x;

// 存储路径的结点跨度
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i;
int level;

2. 搜索当前结点插入位置

serverAssert(!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];
    
    // 如果 score 相等，还需要比较 value 值 【重要！】
    while (x->level[i].forward && 
                   (x->level[i].forward->score < score ||
                            (x->level[i].forward->score == score && 
                                     sdscmp(x->level[i].forward->ele,ele) < 0))) { 
          rank[i] += x->level[i].span; x = x->level[i].forward; 
    }
    
    // 记录 "搜索路径" 
    update[i] = x;
  
}

3. 生成插入结点

level = zslRandomLevel(); 

// 如果随机生成的 level 超过了当前最大 level 需要更新跳跃表的信息 
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,ele);

4. 重排前向指针

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;
} 

/* increment span for untouched levels */ 
for (i = level; i < zsl->level; i++) { 
    update[i]->level[i].span++; 
}

5. 重排后向指针并返回

x->backward = (update[0] == zsl->header) ? NULL : update[0]; 

if (x->level[0].forward) {
    x->level[0].forward->backward = x; 
} else {
    zsl->tail = x;
}

zsl->length++; 

return x;

• 删除实现

删除过程和插入过程实现类似，都需要先找出“搜索路径”，然后对于每个层的相关结点重排前向和后向指针，同时还要注意更新最高层数maxLevel属性。源码实现：

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */ 
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) { 
    int i;
    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;
    } 
  
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL) {
        zsl->level--;
    } 
  
    zsl->length--; 
} 

int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) { 
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x; 
    int i; 
  
    x = zsl->header; 
    for (i = zsl->level-1; i >= 0; i--) { 
        while (x->level[i].forward && 
               (x->level[i].forward->score < score || 
                (x->level[i].forward->score == score && 
                 sdscmp(x->level[i].forward->ele,ele) < 0))) { 
              x = x->level[i].forward; 
        } 
      
        update[i] = x; 
    }
  
    x = x->level[0].forward; 
    if (x && score == x->score && sdscmp(x->ele,ele) == 0) { 
        zslDeleteNode(zsl, x, update); 
        if (!node) {
            zslFreeNode(x);
        } else {
            *node = x;
        } 
        return 1; 
    } 
    return 0; /* not found */ 
}

• 更新实现

我们使用sorted set结构时，调用zadd方法，如果对应的value不存在，那就是插入过程；如果value已经存在，只是调整一下score值，此时需要执行更新过程。

在更新过程中，如果这个新的score值没有带来排序上的变化，就不需要调整位置，直接修改元素的score值即可；如果排序位置改变了，就需要调整位置了。Redis是如何进行调整的呢？源码见晓：把这个元素删除再插入！！！

/* Remove and re-insert when score changed. */ 
if (score != curscore) { 
    zobj->ptr = zzlDelete(zobj->ptr,eptr); 
    zobj->ptr = zzlInsert(zobj->ptr,ele,score); 
    *flags |= ZADD_UPDATED; 
}

往期精彩文章

Redis系列（一）：与Redis的第一次相识

MySQL系列文章