Redis數據類型使用場景及有序集合SortedSet底層實現詳解

redisObject源碼(server.h)

typedef struct redisObject {
    unsigned type:4;//對象類型,用於分辨字元串、列表、
//集合、有序集合、字典,有序集合為REDIS_ZSET
    unsigned encoding:4;//編碼,標識底層數據結構,
//有序集合有REDIS_ENCODING_ZIPLIST(壓縮列表)、REDIS_ENCODING_SKIPLIST(跳錶)
    //記錄鍵最近一次被訪問的時間,長時間不被訪問的對象可被記憶體回收
    unsigned lru:LRU_BITS; 

/* LRU time (relative to global lru_clock) or
 * LFU data (least significant 8 bits frequency
 * and most significant 16 bits access time). */
    int refcount;//引用計數,用於對象記憶體回收,
//當為0時回收記憶體,引用計數可實現不同鍵相同值的共用,
//事實上,redis會初始化創建0到9999個整數對象用於共用,從而節約記憶體
    void *ptr;//指向底層數據結構實例的指針
} robj;

zaddGenericCommand函數

if (server.zset_max_ziplist_entries == 0 ||
server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
        {
            zobj = createZsetObject(); //創建zset
        } else {
            zobj = createZsetZiplistObject();//創建ziplist
        }

Field

含義

zlbytes(uint32_t)

ziplist占用的記憶體位元組數,包括zlbytes本身

zltail(uint32_t)

最後一個entry的offset偏移值

zllen(uint16_t)

數據項entry的個數

entry(變長)

數據項

zlend(uint8_t)

標識ziplist的結束,值為255

Field

含義

prevlen

上一個數據項entry的長度。當長度小於254位元組,則prevlen占1位元組,當長度大於或等於254位元組,則prevlen占5位元組,首位元組值為254,剩下4位元組表示上一entry長度。

encoding

encoding的值依賴於數據entry-data。首位元組的前兩個bit為00、01、10,標識entry-data為字元串,同時表示encoding的長度分別為1、2、5位元組,除前兩個bit，剩下的bit表示字元串長度;前兩個bit為11,表示entry-data為整型,接下來的2 bit表示整數類型。entry-data不同類型及encoding如下:

1)       |00pppppp| - 1 byte,字元串且長度小於等於63位元組(6bit)

2)       |01pppppp|qqqqqqqq| - 2 bytes,字元串且長度小於等於16383位元組(14bit)

3)       |10000000|qqqqqqqq|rrrrrrrr|ssssssss|tttttttt| - 5 bytes,字元串且長度大於等於16384(後面四個位元組表示長度,首位元組的低位6bit設為0)

4)       |11000000| - 1 bytes,len欄位為1位元組,後面的entry-data為整型且類型為int16_t (2 bytes)

5)       |11010000| - 1 bytes, entry-data為整型且類型為int32_t (4 bytes)

6)       |11100000| - 1 bytes, entry-data為整型且類型為int64_t (8 bytes)

7)       |11110000| - 1 bytes, entry-data為整型且占3 bytes

8)       |11111110| - 1 bytes, entry-data為整型且占1 bytes

9)       |1111xxxx| - (with xxxx between 0000 and 1101),xxxx的值從1到13,可用於表示entry-data(1到12),encoding包含entry-data的值,從而不需要entry-data域

10)    |11111111| - 用於標識ziplist的結束

entry-data

具體的數據

ziplist插入邏輯源碼(ziplist.c)

/* Insert item at "p". */

unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {

    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen;

    unsigned int prevlensize, prevlen = 0;

    size_t offset;

    int nextdiff = 0;

    unsigned char encoding = 0;

    long long value = 123456789; /* initialized to avoid warning. Using a value

                                    that is easy to see if for some reason

                                    we use it uninitialized. */

    zlentry tail;

 

    /* Find out prevlen for the entry that is inserted. */

    //插入位置前面一個entry節點占用的位元組數prevlen

    if (p[0] != ZIP_END) {//插入節點不在末尾節點,直接從p的前面位元組讀

        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);

    } else {//插入節點在末尾位置,找到末尾節點

        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);

        if (ptail[0] != ZIP_END) {

            prevlen = zipRawEntryLength(ptail);

        }

    }

    /* See if the entry can be encoded */

    if (zipTryEncoding(s,slen,&value,&encoding)) {//判斷s是否可以轉化為整數,並將整數值和enconding分別存在value和encoding指針

        /* 'encoding' is set to the appropriate integer encoding */

        reqlen = zipIntSize(encoding);//整數值長度

    } else {

        /* 'encoding' is untouched, however zipStoreEntryEncoding will use the

         * string length to figure out how to encode it. */

        reqlen = slen;//字元串長度

    }

 

    /* We need space for both the length of the previous entry and

     * the length of the payload. */

    //得出新插入節點占用的總位元組數reqlen

    reqlen += zipStorePrevEntryLength(NULL,prevlen);

    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

 

    /* When the insert position is not equal to the tail, we need to

     * make sure that the next entry can hold this entry's length in

     * its prevlen field. */

    //插入新節點不在末尾位置,則插入位置p所指向的entry節點的prevlen,

    //值會變成新插入節點的總長度,且prevlen所占用的位元組數可能會變化,

    //nextdiff表示新插入節點下一節點的prevlen需要空間的變化,負值表示變小,

    //正值表示擴大

    int forcelarge = 0;

    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;

    if (nextdiff == -4 && reqlen < 4) {

        nextdiff = 0;

        forcelarge = 1;

    }

 

    /* Store offset because a realloc may change the address of zl. */

    offset = p-zl;

    zl = ziplistResize(zl,curlen+reqlen+nextdiff);//重新分配空間,並將zl的每位元組都填充到新分配的記憶體中

    p = zl+offset;

    //將p後面的數據項進行移動

    /* Apply memory move when necessary and update tail offset. */

    if (p[0] != ZIP_END) {

        /* Subtract one because of the ZIP_END bytes */

        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);

 

        /* Encode this entry's raw length in the next entry. */

        if (forcelarge)//設置下一個節點的prevlen

            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);

        else

            zipStorePrevEntryLength(p+reqlen,reqlen);

 

        /* Update offset for tail */

        ZIPLIST_TAIL_OFFSET(zl) =

            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

 

        /* When the tail contains more than one entry, we need to take

         * "nextdiff" in account as well. Otherwise, a change in the

         * size of prevlen doesn't have an effect on the *tail* offset. */

        zipEntry(p+reqlen, &tail);

        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {

            ZIPLIST_TAIL_OFFSET(zl) =

                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);

        }

    } else {

        /* This element will be the new tail. */

        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);

    }

 

    /* When nextdiff != 0, the raw length of the next entry has changed, so

     * we need to cascade the update throughout the ziplist */

    if (nextdiff != 0) {

        offset = p-zl;

        zl = __ziplistCascadeUpdate(zl,p+reqlen);

        p = zl+offset;

    }

 

    /* Write the entry */

    //將新數據項放入插入位置

    p += zipStorePrevEntryLength(p,prevlen);

    p += zipStoreEntryEncoding(p,encoding,slen);

    if (ZIP_IS_STR(encoding)) {

        memcpy(p,s,slen);

    } else {

        zipSaveInteger(p,value,encoding);

    }

    ZIPLIST_INCR_LENGTH(zl,1);

    return zl;

}

zset定義(server.h)

typedef struct zset {

    dict *dict;//字典

    zskiplist *zsl;//跳錶

} zset;

zslRandomLevel函數(t_zset.c)

int zslRandomLevel(void) {

    //隨機生成節點層數,當第i層節點存在時，第i+1節點存在的概率為ZSKIPLIST_P = 1/4

    //ZSKIPLIST_MAXLEVEL 64,表示節點的最大層數

    int level = 1;

    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))

        level += 1;

    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;

}

zskiplist及zskiplistNode(server.h)

/* ZSETs use a specialized version of Skiplists */

typedef struct zskiplistNode {

    sds ele;//實際存儲的數據

    double score;//分值

    struct zskiplistNode *backward;//後退指針,指向前一個節點

    struct zskiplistLevel {

        struct zskiplistNode *forward;//前進指針,指向下一個節點

        unsigned long span;//跨度,表示該層鏈表的這一節點到下一節點跨越的節點數,用於計算rank

    } level[];//層級數組,每個層級都有到下一個節點的指針和跨度

} zskiplistNode;//跳錶節點

 

typedef struct zskiplist {

    struct zskiplistNode *header, *tail;//跳錶頭節點和尾節點

    unsigned long length;//跳錶元素個數

    int level;//跳錶的最高層數(不包括頭節點,頭節點實際上並不存儲數據)

} zskiplist;

zslInsert(t_zset.c)

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;

    for (i = zsl->level-1; i >= 0; i--) {

        /* store rank that is crossed to reach the insert position */

        //rank[i]初始化為rank[i+1],所以rank[i]-rank[i+1]表示在i層走過的節點數

        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];

        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;

        }

        // 記錄將要和新節點相連接的節點,x表示新節點在i層連接的上一節點

        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. */

    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,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 */

        //rank[0]表示0層鏈表，插入節點x左邊的節點數

        //rank[i]表示i層鏈表,插入節點x左邊的節點數

        //rank[0] - rank[i]+1表示i層鏈表,x前一節點到x的跨度

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

    //在level及之上的每層,update[i]到下一節點的距離由於插入了x節點而加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;

}