Redis有序集內部實現原理分析(二)

Redis技術交流群 481804090

Redis:https://github.com/zwjlpeng/Redis_Deep_Read

本篇博文緊隨上篇Redis有序集內部實現原理分析，在這篇博文里凡出現源碼的地方均以下述src/version.h中定義的Redis版本為主

#define REDIS_VERSION "2.9.11"

在上篇博文Redis有序集內部實現原理分析中，我分析了Redis從什么時候開始支持有序集、跳表的原理、跳表的結構、跳表的查找/插入/刪除的實現,理解了跳表的基本結構，理解Redis中有序集的實現就不難了，因為Redis有序集的實現也是以跳表作為基礎的底層數據結構，選擇這種數據結構，不僅僅是因為簡單，更多的是因為性能。

Redis中跳表的基本數據結構定義如下，與基本跳表數據結構相比，在Redis中實現的跳表其特點是不僅有前向指針，也存在后向指針，而且在前向指針的結構中存在span跨度字段，這個跨度字段的出現有助於快速計算元素在整個集合中的排名

//定義跳表的基本數據節點
typedef struct zskiplistNode {
    robj *obj; // zset value
    double score;// zset 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;

//有序集數據結構
typedef struct zset {
    dict *dict;//字典存放value,以value為key
    zskiplist *zsl;
} zset;

將如上數據結構轉化成更形式化的圖形表示，如下圖所示

 

在上圖中，可以看到header指針指向的是一個具有固定層級(32層)的表頭節點,為什么定義成32,是因為定義成32層理論上對於2^32-1個元素的查詢最優，而2^32=4294967296個元素，對於絕大多數的應用來說，已經足夠了，所以就定義成了32層,到於為什么查詢最優，你可以將其想像成一個32層的完全二叉排序樹，算算這個樹中節點的數量

Redis中有序集另一個值得注意的地方就是當Score相同的時候，是如何存儲的，當集合中兩個值的Score相同，這時在跳表中存儲會比較這兩個值，對這兩個值按字典排序存儲在跳表結構中

有了上述的數據結構相關的基礎知識，來看看Redis對zskiplist/zskiplistNode的相關操作,源碼如下所示(源碼均出自t_zset.c)

創建跳表結構的源碼

//#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    //分配內存
    zsl = zmalloc(sizeof(*zsl));
    zsl->level = 1;//默認層級為1
    zsl->length = 0;//跳表長度設置為0
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        //因為沒有任何元素，將表頭節點的前向指針均設置為0
        zsl->header->level[j].forward = NULL;
        //將表頭節點前向指針結構中的跨度字段均設為0
        zsl->header->level[j].span = 0;
    }
    //表頭后向指針設置成0
    zsl->header->backward = NULL;
    //表尾節點設置成NULL
    zsl->tail = NULL;
    return zsl;
}

在上述代碼中調用了zslCreateNode這個函數,函數的源碼如下所示=

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

執行完上述代碼之后會創建如下圖所示的跳表結構

 

創建了跳表的基本結構，下面就是插入操作了，Redis中源碼如下所示

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x; //update[32]
    unsigned int rank[ZSKIPLIST_MAXLEVEL];//rank[32]
    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];
        while (x->level[i].forward &&
            (x->level[i].forward->score < score || //以下是得分相同的情況下，比較value的字典排序
                (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;
    }
    //產生隨機層數
    level = zslRandomLevel();
    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;
}

上述源碼中，有一個產生隨機層數的函數，源代碼如下所示:

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

圖形化的形式描述如下圖所示:

理解了插入操作，其他查詢，刪除，求范圍操作基本上類似，此處忽略...