跳跃表skiplist

简介

你一定比较好奇Redis里面的 sorted set 是怎么实现的，底层到底是什么？它的排序功能就是用到了这个skiplist-跳跃表。

什么是跳跃表？

跳跃表可以看做是链表的一个变种，一个多层顺序链表层级结构组成。它是一种顺序查找数据结构。

它是由William Pugh发明的，最早出现于他在1990年发表的论文《Skip Lists: A Probabilistic Alternative to Balanced Trees。

看一张图，能形象的记住

来自：http://prismoskills.appspot.com/lessons/Algorithms/Chapter_04_-_Skip_Lists.jsp

它有着堪比红黑树的效率，但是比红黑树的操作简单多了。

这里有一份跳跃表各种概念介绍，很详细：https://www.iteye.com/blog/kenby-1187303

Redis中的跳跃表

Redis中的数据结构定义和操作API，编写的非常简洁明了，十分适合阅读。

Redis 中就使用了这种数据结构，它用于zset的排序操作，数据结构定义如下：

数据结构定义:

// src/redis.h  redis3.0 老版本结构简单

typedef struct zskiplistNode {
    robj *obj;                        // 对象成员
    double score;                      //分数
    struct zskiplistNode *backward;    //向后的指针
    struct zskiplistLevel {            //水平层级
        struct zskiplistNode *forward;  //向前指针
        unsigned int span;             // 跨度(redis中用于计算元素排名)
    } level[];
} zskiplistNode;

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

层 level：
跳跃表的节点的level数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层加快访问其他节点的速度。

层的高度怎么确定呢？
程序根据幂次定律随机生成一个介于1和32之间的值作为level数组的大小。

向前指针 forward：
用于访问前面的元素

跨度 span：
用于记录2个节点之间的距离。 如果指向NULL，那么跨度就为0。

分数 score：
是一个double的浮点数，跳跃表中的所有节点都是按照分值从小到大来排序的。

成员对象 obj：
是一个指针，指向一个字符串对象，字符串对象保存在SDS中。

在跳跃表中，各个节点保存的成员对象必须唯一，但是多个节点保存的分值却可以相同。分值按从小到大来排序，成员对象的分值较小的排在前面。

操作的API:

创建新的跳跃表

// src/t_zset.c redis3.0

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;

    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;
    zsl->length = 0;
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}

创建跳跃表节点Node

zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
    zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    zn->score = score;
    zn->obj = obj;
    return zn;
}

释放跳跃表以及表中包含的节点

// src/t_zset.c redis3.0

void zslFree(zskiplist *zsl) {
    zskiplistNode *node = zsl->header->level[0].forward, *next;

    zfree(zsl->header);
    while(node) {
        next = node->level[0].forward;
        zslFreeNode(node);
        node = next;
    }
    zfree(zsl);
}

插入节点到跳跃表中

// src/t_zset.c redis3.0

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

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    
    // 记录每一层插入节点的前面一个节点在skiplist中的排名
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    redisAssert(!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比较，相等即根据sds的字符串字典序比较
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                (x->level[i].forward->score == score &&
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;// 每一个层级的待插入位置，即当前层最后一个小于x的点
    }
    /* we assume the key is not already inside, since we allow duplicated
     * scores, and the re-insertion of score and redis object should never
     * happen since the caller of zslInsert() should test in the hash table
     * if the element is already inside or not. */
    level = zslRandomLevel();
    // 如果计算出来的层级比当前层级高，则重设超出zsl原来层级的指针
    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,obj);
    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 */
     /* 自增没有到达的层级的span */
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }

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

删除节点

// src/t_zset.c redis3.0

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