目录

跳跃表(skiplist)是一种有序数据结构, 它通过在每个节点中维持多个指向其他节点的指针, 从而达到快速访问节点的目的。

跳跃表支持平均 O(\log N) 最坏 O(N) 复杂度的节点查找, 还可以通过顺序性操作来批量处理节点。

在大部分情况下, 跳跃表的效率可以和平衡树相媲美, 并且因为跳跃表的实现比平衡树要来得更为简单, 所以有不少程序都使用跳跃表来代替平衡树。

Redis 使用跳跃表作为有序集合键的底层实现之一: 如果一个有序集合包含的元素数量比较多, 又或者有序集合中元素的成员(member)是比较长的字符串时, Redis 就会使用跳跃表来作为有序集合键的底层实现。

和链表、字典等数据结构被广泛地应用在 Redis 内部不同, Redis 只在两个地方用到了跳跃表, 一个是实现有序集合键, 另一个是在集群节点中用作内部数据结构, 除此之外, 跳跃表在 Redis 里面没有其他用途。

先看一下维基百科对跳跃表的图示: 跳跃表

从图中可以看到, 跳跃表主要由以下部分构成:

  • 表头(head):负责维护跳跃表的节点指针。
  • 跳跃表节点:保存着元素值,以及多个层。
  • 层:保存着指向其他元素的指针。高层的指针越过的元素数量大于等于低层的指针,为了提高查找的效率,程序总是从高层先开始访问,然后随着元素值范围的缩小,慢慢降低层次。
  • 表尾:全部由 NULL 组成,表示跳跃表的末尾。

跳跃表结构定义

跳跃表的结构体定义在server.h文件中。其中包括跳跃表节点zskiplistNode和跳跃表zskiplist两个结构体。

/* ZSETs use a specialized version of Skiplists */
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, *tail;   // 头尾结点
    unsigned long length;    // 总的结点数
    int level;    // 总的层数
} zskiplist;

每次创建一个新跳跃表节点的时候, 程序都根据幂次定律 (power law,越大的数出现的概率越小) 随机生成一个介于 1 和 32 之间的值作为 level 数组的大小, 这个大小就是层的“高度”。

总的层数保存在zskiplist的level参数中, 另外每个节点保存了各自层中的指针以及这一层的跨度。

跳跃表操作

创建跳跃表

/* Create a skiplist node with the specified number of levels.
 * The SDS string 'ele' is referenced by the node after the call. */
 // 创建跳跃表结点
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
    // 开辟内存,根据传入的层数设置大小
    zskiplistNode *zn =
        zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    // 赋值跳跃表结点分值
    zn->score = score;
    // 赋值跳跃表结点对象
    zn->ele = ele;
    return zn;
}

/* 创建跳跃表 */
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 申请内存
    zsl = zmalloc(sizeof(*zsl));
    // 设置层数为1
    zsl->level = 1;
    // 总的结点数为0
    zsl->length = 0;
    // ZSKIPLIST_MAXLEVEL=32, 头结点,设置最大层数,分值为0, 具体对象为NULL
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    // 循环设置头结点的每一层的前向指针为NULL,跨度为0
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    // 头结点的后向指针为NULL
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

创建跳跃表的时候设置层数为1, 只有一个头结点,头结点保存了最大的层数,同时所有的前向指针都为NULL。

释放整个跳跃表

/* Free the specified skiplist node. The referenced SDS string representation
 * of the element is freed too, unless node->ele is set to NULL before calling
 * this function. */
void zslFreeNode(zskiplistNode *node) {
    sdsfree(node->ele);
    zfree(node);
}

/* 释放整个跳跃表. */
void zslFree(zskiplist *zsl) {
    // 从最底层的level[0]依次遍历,释放
    zskiplistNode *node = zsl->header->level[0].forward, *next;

    zfree(zsl->header);
    // 如果存在后续结点
    while(node) {
        next = node->level[0].forward;
        // 释放结点
        zslFreeNode(node);
        node = next;
    }
    zfree(zsl);
}

跳跃表插入元素

/* Insert a new node in the skiplist. Assumes the element does not already
 * exist (up to the caller to enforce that). The skiplist takes ownership
 * of the passed SDS string 'ele'. */
 // 跳跃表插入元素
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));    // 判断是否为数字
    x = zsl->header;
    // 从最高的level, 也即跨度最大的level开始查找结点
    for (i = zsl->level-1; i >= 0; i--) {
        /* store rank that is crossed to reach the insert position */
        // 当前是否是最高层, 如果是最高层,rank[i]=0,否则,复制上一层的数值
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 如果当前结点的score值小于传入的score 或者 当前score相等,但是结点的对象不相等
        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]
            rank[i] += x->level[i].span;
            // 在当前层中向前查找
            x = x->level[i].forward;
        }
        // 当前层位于插入位置前的结点x放入update数组
        update[i] = x;
    }
    /* we assume the element is not already inside, since we allow duplicated
     * scores, reinserting the same element should never happen since the
     * caller of zslInsert() should test in the hash table if the element is
     * already inside or not. */
    // 随机生成小于32的层数
    level = zslRandomLevel();
    // 如果生成的层数大于当前的层数
    if (level > zsl->level) {
        for (i = zsl->level; i < level; i++) {
            // 设定rank数组中大于原level层以上的值为0
            // 同时设定update数组大于原level层以上的数据
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        zsl->level = level;
    }
    // 创建层数为level的新结点
    x = zslCreateNode(level,score,ele);
    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++;
    }

    // 根据最低层的前序结点是否是header结点来设置当前新结点的向后指针
    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;
}

首先看图示,如果想要插入score=5的结点(redis中允许score值重复),那么首选需要找到score=5的结点,查找的顺序为:

logo

从最大的层向后查找,如果当前层后边没有值了,并且当前结点的值小于要找的值,就查找下一层结点;如果下一个结点的值大于要找的值,也会到下一层结点继续查找。

找到对应的位置执行插入操作后,需要为新结点设置层数,那么设置多少层合适呢,这边直接采用了一个随机数。随机数生成了多少层,当前新结点的层数就设置多少层。

如果新层数小于原来的层数,只需要重新设置前序后置结点的指针和跨度就行;如果新层数大于原来的层数,就需要额外设置新的更高的层。

那么插入结点之后,如何修改前序和后置结点的指针和跨度呢。这边用了两个数组来记录,分别是update和rank。update用来记录每一层中插入位置的前序结点,到时候根据这个前序结点设置每一层的指针调整。 rank用来记录每一层到新结点的跨度,插入新结点之后,根据rank数组中记录跨度更新前置结点的跨度值。

删除跳跃表结点

 /* 删除结点,如果结点存在并删除,返回1, 否则返回0
    参数中node如果是空的,则结点确实被删除; 如果非空,只是把结点从链表上摘下来,返回指针给node*/
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;
    }
    /* We may have multiple elements with the same score, what we need
     * is to find the element with both the right score and object. */
    // 由于允许存在相同的score,需要在score和ele都满足的条件下才能删除
    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 */
}

具体的结点删除操作:

/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
// update数组保存了所有层上要删除结点的前置结点
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 {
            // 否则,只需要将前置结点的跨度减1即可,因为少了一个结点啊
            update[i]->level[i].span -= 1;
        }
    }
    // 修改backward指针,需要考虑x是否为尾节点
    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--;
}