Redis 2.8.9源碼 - 跳錶的實現

本文爲做者原創，轉載請註明出處：http://my.oschina.net/fuckphp/blog/280168

本文代碼能夠在 src/redis.h 和 src/t_zset.c 兩個文件中找到，關於跳錶相關的api請參考另一篇文章   Redis 2.8.9源碼 - 跳錶操做 操做函數頭整理，並註釋做用和參數說明

Redis中用zset數據結構，用來存放一組有序的數據集，能夠實現排行榜等功能，zset的內部實現使用的是一個叫作 跳錶（skiplist）的數據結構，這個數據結構是由 William Pugh（《Skip lists: a probabilistic alternative to balanced trees》） 發明的一個性能上能夠和平衡術媲美的數據結構，不熟悉此數據結構的能夠見文章結尾的參考資料。

首先看一下實現跳錶的數據結構

跳錶節點（以下代碼定義在 src/redis.h 中）：

typedef struct zskiplistNode {
    //跳錶節點中key的對象
    robj *obj;
    //跳錶節點中存儲的分值
    double score;
    //跳錶節點中 存放的前驅節點
    struct zskiplistNode *backward;
    //保存每一個節點中每一層指向的後繼節點數組
    struct zskiplistLevel {
        //存放每一層的節點指針
        struct zskiplistNode *forward;
        //當前節點與後繼節點跨過的節點數量
        unsigned int span;
    } level[];
} zskiplistNode;

跳錶結構（以下代碼定義在 src/redis.h 中）：

typedef struct zskiplist {
    //指向跳錶節點的頭尾指針
    struct zskiplistNode *header, *tail;
    //跳錶節點的個數
    unsigned long length;
    //跳錶節點最大層數
    int level;
} zskiplist;

而後咱們看一下跳錶的建立

建立跳錶節點（以下代碼定義在 src/t_zset.c 中）：

zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
    //根據當前節點 的層高來 分配 不一樣大小的內存
    //內存大小爲 結構自己 +　層數　* 每一層的佔用的空間
    zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    //設置分值
    zn->score = score;
    //設置key對象
    zn->obj = obj; 
    //返回建立好的 節點指針
    return zn;
}

建立跳錶（以下代碼定義在 src/t_zset.c 中）：

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    //分配內存空間
    zsl = zmalloc(sizeof(*zsl));
    //設置默認層高爲1 （每一個節點都有一個層高爲1 指向後繼接節點的指針）
    zsl->level = 1; 
    //節點數量
    zsl->length = 0; 
    //設置頭結點 頭結點高度默認爲 32
    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; 
}

跳錶的插入（以下代碼定義在 src/t_zset.c 中）：

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    //定義 節點數組 update用戶保存插入節點每一層的前驅節點
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    //rank 數組 用於保存 插入節點插入位置在每一層中跨國的節點數
    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 */
        //若是是最高層，則標示擴過0個 節點，不然 記錄跨過的節點數 與上一層相同
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        //經過比較 若是插入分值小於右邊節點就繼續 向右遍歷，並計算每一層的跨度，若是大於右邊則下降一層
        //若是相等則 經過key對象進行大小比較
        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數組中
        update[i] = 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();
    //若是層高 比 當前層高 要高，則將header的指針指向 超出當前層高的部分
    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;

        //計算您當前節點與後繼節點的跨過的節點數
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    
    //插入位置的節點 與 新插入的節點 之間跨過的 節點數
    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;
}

跳錶的遍歷方式與 插入方式相同，都是從高層開始遍歷，逐漸下降層數。

Redis2.8.9源碼   src/redis.h   src/t_zset.c

Redis 設計與實現（初版）

一個C語言實現的跳錶結構

Skiplist 跳錶