Redis源碼閱讀筆記-對象及其類型和編碼

總結之《Redis設計與實現》

對象

Redis中是使用對象來即是數據庫中的鍵和值。

結構

// server.h
...
#define LRU_BITS 24
...

typedef struct redisObject {
    unsigned type:4;        
    unsigned encoding:4;
    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;
    void *ptr;
} robj;

• type: 4 bits, 表示是什麼類型的Redis Object。
• encoding: 4 bits, 類型的編碼，是指定使用type的編碼方式。
• *ptr: 指向底層實現數據結構的指針。
• lru: 24 bits, 記錄了該對象最後一次被命令程序訪問的時間。
• refcount: 引用計數。

類型

對象	對象type的屬性值	TYPE命令的輸出
字符串	OBJ_STRING 0	"string"
列表	OBJ_LIST 1	"list"
集合	OBJ_SET 2	"set"
有序集合	OBJ_ZSET 3	"zset"
哈希	OBJ_HASH 4	"hash"

類型的編碼

編碼常量	對應底層數據結構
OBJ_ENCODING_RAW 0	簡單動態字符串，只有string類型纔會使用這個encoding值
OBJ_ENCODING_INT 1	long類型的整數
OBJ_ENCODING_HT 2	字典dict
OBJ_ENCODING_ZIPMAP 3	舊類，再也不使用
OBJ_ENCODING_LINKEDLIST 4	雙端列表，已再也不用
OBJ_ENCODING_ZIPLIST 5	壓縮列表ziplist
OBJ_ENCODING_INTSET 6	整數集合intset
OBJ_ENCODING_SKIPLIST 7	跳躍表skiplist，用於有序集合
OBJ_ENCODING_EMBSTR 8	表示一種與robj嵌入的特殊sds字符串
OBJ_ENCODING_QUICKLIST 9	quicklist快速列表

字符串對象

類型	編碼	對象
OBJ_STRING	OBJ_ENCODING_RAW 0	使用SDS字符串實現的字符串對象
OBJ_STRING	OBJ_ENCODING_INT 1	使用整數值實現的字符串對象（整數值類型爲long）
OBJ_STRING	OBJ_ENCODING_EMBSTR 8	使用SDS字符串實現的字符串對象，主要區別爲在內存中字符串與對象相鄰

OBJ_ENCODING_INT

在robj中，有一個屬性refcount是用來保存引用計數，對於字符串對象，Redis會在初始化服務器時，建立1萬個字符串對象，包含了 0 ~ 9999 的整數字，當須要使用到這些整數字符串對象，服務器就會使用這些共享對象，而不須要新建整數對象。

OBJ_ENCODING_EMBSTR 與 OBJ_ENCODING_RAW的區別

若是對象中保存的字符串長度小於OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44將會使用OBJ_ENCODING_EMBSTR編碼。

//object.c
/* Create a string object with EMBSTR encoding if it is smaller than
 * OBJ_ENCODING_EMBSTR_SIZE_LIMIT, otherwise the RAW encoding is
 * used.
 *
 * The current limit of 39 is chosen so that the biggest string object
 * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44
robj *createStringObject(const char *ptr, size_t len) {
    if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)
        // 若是長度小於OBJ_ENCODING_EMBSTR_SIZE_LIMIT，將會調用createEmbeddedStringObject()建立 OBJ_ENCODING_EMBSTR 編碼的字符串對象
        return createEmbeddedStringObject(ptr,len);
    else
        // 若是長度大於OBJ_ENCODING_EMBSTR_SIZE_LIMIT，將會調用createRawStringObject()建立 OBJ_ENCODING_RAW 編碼的字符串對象
        return createRawStringObject(ptr,len);
}

OBJ_ENCODING_EMBSTR編碼是用於專門保存短字符串的優化方式，同樣會使用SDS結構來保存字符串。但OBJ_ENCODING_RAW編碼會調用兩次內存分配函數來分別建立robj和SDS字符串。而OBJ_ENCODING_EMBSTR編碼則經過調用一次內存分配函數來分配一個連續的空間，空間中依次包含robj和SDS字符串。

OBJ_ENCODING_EMBSTR優勢:

• 下降內存分配/釋放的次數。
• 減小內存碎片，使得OBJ_ENCODING_EMBSTR編碼的字符串更好得利用SDS緩存帶來的優點。
• 當對OBJ_ENCODING_EMBSTR編碼的字符串對象中字符串作任何改變，都會使其轉換成OBJ_ENCODING_RAW編碼。

建立函數代碼

• robj *createStringObjectFromLongLong(long long value)建立OBJ_ENCODING_INT編碼的字符串對象:

  // object.c
  
  	robj *createStringObjectFromLongLong(long long value) {
  	    robj *o;
  	    // 若是值在 0 ~ 9999 之間，則返回共享的整數字符串對象
  	    if (value >= 0 && value < OBJ_SHARED_INTEGERS) {
  	        // 若是值在 0 ~ 9999 之間，則返回共享的整數字符串對象
  	        incrRefCount(shared.integers[value]);
  	        o = shared.integers[value];
  	    } else {
  
  	        if (value >= LONG_MIN && value <= LONG_MAX) {
  	            // 若是值是在 long 的範圍內，則直接保存long類型的整數值
  	            // 來建立整數字符串對象
  	            o = createObject(OBJ_STRING, NULL);
  	            o->encoding = OBJ_ENCODING_INT;
  	            o->ptr = (void*)((long)value);
  	        } else {
  	            // 若是值是在 long 的範圍外，則使用SDS字符串來保存整數
  	            // 來建立整數字符串對象
  	            o = createObject(OBJ_STRING,sdsfromlonglong(value));
  	        }
  	    }
  	    return o;
  	}

• robj *createEmbeddedStringObject(const char *ptr, size_t len) 建立OBJ_ENCODING_EMBSTR編碼的字符串對象:

  /* Create a string object with encoding OBJ_ENCODING_EMBSTR, that is
  	 * an object where the sds string is actually an unmodifiable string
  	 * allocated in the same chunk as the object itself. */
  	robj *createEmbeddedStringObject(const char *ptr, size_t len) {
  	    // 申請robj、sdshdr和字符長度之和的內存
  	    robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr8)+len+1);
  	    // sdshdr8 緊接着 robj
  	    struct sdshdr8 *sh = (void*)(o+1);
  
  	    // 初始化robj的屬性值
  	    o->type = OBJ_STRING;
  	    o->encoding = OBJ_ENCODING_EMBSTR;
  	    // 指向sds字符串結構
  	    o->ptr = sh+1;
  	    o->refcount = 1;
  	    // 設置lru訪問時間
  	    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
  	        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
  	    } else {
  	        o->lru = LRU_CLOCK();
  	    }
  
  		 // sds字符串初始化
  	    sh->len = len;
  	    sh->alloc = len;
  	    sh->flags = SDS_TYPE_8;
  	    if (ptr) {
  	        memcpy(sh->buf,ptr,len);
  	        sh->buf[len] = '\0';
  	    } else {
  	        memset(sh->buf,0,len+1);
  	    }
  	    return o;
  	}

• robj *createRawStringObject(const char *ptr, size_t len)建立OBJ_ENCODING_RAW編碼的字符串對象:

  /* Create a string object with encoding OBJ_ENCODING_RAW, that is a plain
  	 * string object where o->ptr points to a proper sds string. */
  	robj *createRawStringObject(const char *ptr, size_t len) {
  	    // 直接調用createObject()建立字符串對象
  	    return createObject(OBJ_STRING, sdsnewlen(ptr,len));
  	}
  
  	robj *createObject(int type, void *ptr) {
  	    // 申請內存
  	    robj *o = zmalloc(sizeof(*o));
  	    // 初始化部分屬性
  	    o->type = type;
  	    o->encoding = OBJ_ENCODING_RAW;
  	    o->ptr = ptr;
  	    o->refcount = 1;
  
  	    // 設置LRU程序訪問時間
  	    /* Set the LRU to the current lruclock (minutes resolution), or
  	     * alternatively the LFU counter. */
  	    if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
  	        o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
  	    } else {
  	        o->lru = LRU_CLOCK();
  	    }
  	    return o;
  	}

列表對象

類型	編碼	對象
OBJ_LIST	OBJ_ENCODING_ZIPLIST 5	使用壓縮列表建立的列表對象(在Redis 4.0.11代碼中看，彷佛沒有使用該編碼的列表對象了)
OBJ_LIST	OBJ_ENCODING_QUICKLIST 9	使用快速列表建立的列表對象

在Redis 4.0.11代碼中看，列表對象的底層實現，都採用快速列表(quicklist)實現了。

部分功能代碼

• robj *createQuicklistObject(void) 建立一個空的編碼爲OBJ_ENCODING_QUICKLIST的列表對象:

  // object.c
  	robj *createQuicklistObject(void) {
  	    // 建立一個空的快速鏈表
  	    quicklist *l = quicklistCreate();
  	    // 傳入快速列表，建立列表對象，而後將編碼指定爲快速列表
  	    robj *o = createObject(OBJ_LIST,l);
  	    o->encoding = OBJ_ENCODING_QUICKLIST;
  	    return o;
  	}

• void listTypePush(robj *subject, robj *value, int where)將value添加到列對象subject中，where控制是插入隊列尾仍是隊列頭:

  // t_list.c
  
  	void listTypePush(robj *subject, robj *value, int where) {
  	    // 判斷subject是否爲隊列(4.0.11中全部隊列的實現均爲快速列表)
  	    if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
  	        // where 爲 0 表示列表頭，不然表示列表尾
  	        int pos = (where == LIST_HEAD) ? QUICKLIST_HEAD : QUICKLIST_TAIL;
  
  	        // 將value解碼
  	        // 其實是判斷value是哪一種編碼的字符串，若是是OBJ_ENCODING_INT類型的字符串
  	        // 則將其 整型值 轉成一個sds字符串類型，並返回一個新的obj對象
  	        // 若是是 OBJ_ENCODING_EMBSTR 或 OBJ_ENCODING_RAW，則只是對value的引用值+1
  	        value = getDecodedObject(value);
  	        size_t len = sdslen(value->ptr);
  	        // 將value的值加入到快速列表中(頭或尾)
  	        quicklistPush(subject->ptr, value->ptr, len, pos);
  	        // 判斷value是否能夠被釋放
  	        // 若是value的refcount == 1, 將會調用對應釋放函數
  	        // 若是value的refcount > 1, 則將其值減1
  	        decrRefCount(value);
  	    } else {
  	        serverPanic("Unknown list encoding");
  	    }
  	}

• robj *listTypePop(robj *subject, int where) 在隊列subject中從where指定的頭或者尾中彈出(pop)元素項，並返回:

  robj *listTypePop(robj *subject, int where) {
  	    long long vlong;
  	    robj *value = NULL;
  
  	    // where 爲 0 表示列表頭，不然表示列表尾
  	    int ql_where = where == LIST_HEAD ? QUICKLIST_HEAD : QUICKLIST_TAIL;
  	    // 檢查subject的編碼是否爲快速隊列
  	    if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
  	        // 在快速隊列中彈出一個元素項，並調用在快速列表的函數中調用listPopSaver()將返回值封裝成robj
  	        // 若是元素項的值是(long long類型)整數，則寫在vlong中，這時候須要本身將其組裝成robj
  	        if (quicklistPopCustom(subject->ptr, ql_where, (unsigned char **)&value,
  	                               NULL, &vlong, listPopSaver)) {
  	            if (!value)
  	                // 彈出的元素項是long long整數，封裝成robj
  	                value = createStringObjectFromLongLong(vlong);
  	        }
  	    } else {
  	        serverPanic("Unknown list encoding");
  	    }
  	    return value;
  	}
  
  	void *listPopSaver(unsigned char *data, unsigned int sz) {
  	    return createStringObject((char*)data,sz);
  	}

哈希對象

哈希對象的底層編碼是壓縮列表 (ziplist)和字典(dict，hashtable)。

類型	編碼	對象
OBJ_HASH	OBJ_ENCODING_ZIPLIST 5	壓縮列表實現的哈希對象
OBJ_HASH	OBJ_ENCODING_HT 2	字典(哈希表)實現的哈希對象

編碼轉換

在 Redis 4.0.11 中，當知足如下兩條件之一時，OBJ_ENCODING_ZIPLIST編碼的哈希對象，會轉換成OBJ_ENCODING_HT編碼的哈希對象:

• 當哈希對象中鍵值對的數量大於OBJ_HASH_MAX_ZIPLIST_ENTRIES 512(該值能夠經過配置hash-max-ziplist-entries改變)時，將進行編碼轉換。
• 當哈希對象中的鍵字符串或者值字符串大於OBJ_HASH_MAX_ZIPLIST_VALUE 64(該值能夠經過配置hash-max-ziplist-value改變)時，將進行編碼轉換。

默認狀況下，Redis會使用OBJ_ENCODING_ZIPLIST編碼，直至以上兩種狀況的其中一種發生。

PS: 從代碼上看，目前只支持OBJ_ENCODING_ZIPLIST -> OBJ_ENCODING_HT，不支持OBJ_ENCODING_HT -> OBJ_ENCODING_ZIPLIST。

部分功能代碼

• robj *createHashObject(void)建立一個空的哈希對象:

  // object.c
  
  	robj *createHashObject(void) {
  	    // 新建一個壓縮列表
  	    unsigned char *zl = ziplistNew();
  	    // 建立一個哈希對象，並指定編碼爲壓縮列表
  	    robj *o = createObject(OBJ_HASH, zl);
  	    o->encoding = OBJ_ENCODING_ZIPLIST;
  	    return o;
  	}

• int hashTypeSet(robj *o, sds field, sds value, int flags) 將鍵field,值value添加到哈希對象o中:

  // t_hash.c
  
  	/* Add a new field, overwrite the old with the new value if it already exists.
  	 * Return 0 on insert and 1 on update.
  	 *
  	 * By default, the key and value SDS strings are copied if needed, so the
  	 * caller retains ownership of the strings passed. However this behavior
  	 * can be effected by passing appropriate flags (possibly bitwise OR-ed):
  	 *
  	 * HASH_SET_TAKE_FIELD -- The SDS field ownership passes to the function.
  	 * HASH_SET_TAKE_VALUE -- The SDS value ownership passes to the function.
  	 *
  	 * When the flags are used the caller does not need to release the passed
  	 * SDS string(s). It's up to the function to use the string to create a new
  	 * entry or to free the SDS string before returning to the caller.
  	 *
  	 * HASH_SET_COPY corresponds to no flags passed, and means the default
  	 * semantics of copying the values if needed.
  	 *
  	 */
  	#define HASH_SET_TAKE_FIELD (1<<0)
  	#define HASH_SET_TAKE_VALUE (1<<1)
  	#define HASH_SET_COPY 0
  	// 將鍵`field`,值`value`添加到哈希對象`o`中，若是鍵`field`已經存在，將會用新值`value`覆蓋舊值
  	// flags的做用是標誌是否要在函數中釋放`filed`或者`value`
  	int hashTypeSet(robj *o, sds field, sds value, int flags) {
  	    int update = 0;
  
  	    if (o->encoding == OBJ_ENCODING_ZIPLIST) {
  	        // 若是hash對象是 壓縮列表
  	        unsigned char *zl, *fptr, *vptr;
  
  	        // zl指向壓縮列表
  	        zl = o->ptr;
  	        // 獲取壓縮列表中頭元素向的位置
  	        fptr = ziplistIndex(zl, ZIPLIST_HEAD);
  	        if (fptr != NULL) {
  	            // 檢查壓縮列表中，是否已經包含鍵field
  	            fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1);
  	            if (fptr != NULL) {
  	                // 若是鍵field已經存在，則在如下操做中更新對應的value
  	                /* Grab pointer to the value (fptr points to the field) */
  	                // vprt指向對應的值位置
  	                vptr = ziplistNext(zl, fptr);
  	                serverAssert(vptr != NULL);
  	                // 將更新標誌位置爲1
  	                update = 1;
  
  	                // 刪除舊值
  	                /* Delete value */
  	                zl = ziplistDelete(zl, &vptr);
  	                // 添加新值
  	                /* Insert new value */
  	                zl = ziplistInsert(zl, vptr, (unsigned char*)value,
  	                        sdslen(value));
  	            }
  	        }
  	        // 若是標誌位 爲 0，表示filed不在壓縮隊列中，須要插入
  	        // 若是標誌位 爲 1，表示已經更新，無須操做
  	        if (!update) {
  	            /* Push new field/value pair onto the tail of the ziplist */
  	            zl = ziplistPush(zl, (unsigned char*)field, sdslen(field),
  	                    ZIPLIST_TAIL);
  	            zl = ziplistPush(zl, (unsigned char*)value, sdslen(value),
  	                    ZIPLIST_TAIL);
  	        }
  	        o->ptr = zl;
  
  	        /* Check if the ziplist needs to be converted to a hash table */
  	        // 檢查鍵值對的數量，看是否須要將底層實現轉換編碼爲字典(哈希表)
  	        if (hashTypeLength(o) > server.hash_max_ziplist_entries)
  	            hashTypeConvert(o, OBJ_ENCODING_HT);
  	    } else if (o->encoding == OBJ_ENCODING_HT) {
  	        // 若是hash對象的編碼爲字典(哈希表)
  
  	        // 查找鍵filed是否已經存在，若是有這返回de
  	        dictEntry *de = dictFind(o->ptr,field);
  	        if (de) {
  	            // 值已經存在，把舊值釋放掉
  	            sdsfree(dictGetVal(de));
  
  	            // 將新值value寫入，經過flags判斷是否須要釋放value
  	            if (flags & HASH_SET_TAKE_VALUE) {
  	                dictGetVal(de) = value;
  	                value = NULL;
  	            } else {
  	                dictGetVal(de) = sdsdup(value);
  	            }
  	            update = 1;
  	        } else {
  	            // 插入新鍵值對
  
  	            sds f,v;
  	             // 經過flags判斷是否須要釋放field
  	            if (flags & HASH_SET_TAKE_FIELD) {
  	                f = field;
  	                field = NULL;
  	            } else {
  	                f = sdsdup(field);
  	            }
  
  	            // 經過flags判斷是否須要釋放value
  	            if (flags & HASH_SET_TAKE_VALUE) {
  	                v = value;
  	                value = NULL;
  	            } else {
  	                v = sdsdup(value);
  	            }
  	            // 往字典添加新鍵值對
  	            dictAdd(o->ptr,f,v);
  	        }
  	    } else {
  	        serverPanic("Unknown hash encoding");
  	    }
  
  	    /* Free SDS strings we did not referenced elsewhere if the flags
  	     * want this function to be responsible. */
  	    // 經過flags判斷
  	    // 釋放掉對應的failed或者value
  	    if (flags & HASH_SET_TAKE_FIELD && field) sdsfree(field);
  	    if (flags & HASH_SET_TAKE_VALUE && value) sdsfree(value);
  	    return update;
  	}

集合對象

集合對象的底層編碼是字典(dict，hashtable)和整數集合(intset)。

類型	編碼	對象
OBJ_SET	OBJ_ENCODING_HT 2	字典實現的集合對象
OBJ_SET	OBJ_ENCODING_INTSET 6	整數集合實現的集合對象

編碼轉換

在 Redis 4.0.11 中，當知足如下兩條件之一時，OBJ_ENCODING_INTSET編碼的集合對象，會轉換成OBJ_ENCODING_HT編碼的集合對象:

• 當集合中保存的元素，有1個不是整數時，OBJ_ENCODING_INTSET會轉變爲OBJ_ENCODING_HT。
• 當集合中保存的元素數量，超過OBJ_SET_MAX_INTSET_ENTRIES 512(能夠經過設置set-max-intset-entries改變數值)時，OBJ_ENCODING_INTSET會轉變爲OBJ_ENCODING_HT。

PS: 從代碼上看，目前只支持OBJ_ENCODING_INTSET -> OBJ_ENCODING_HT，不支持OBJ_ENCODING_HT -> OBJ_ENCODING_INTSET。

部分功能代碼

• robj *setTypeCreate(sds value) 經過值值value得類型來判斷建立一個集合對象，這個方法並不會將值value放入到集合中:

  robj *setTypeCreate(sds value) {
  	    if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
  	        // 判斷值'value'是否爲一個整數，若是是，則使用整數集合編碼的集合對象
  	        return createIntsetObject();
  	    // 'value'值不是整數，使用字典編碼的集合對象
  	    return createSetObject();
  	}

• int setTypeAdd(robj *subject, sds value) 將值value添加到集合subject中，若是值已經存在，函數返回0，不然返回1:

  /* Add the specified value into a set.
  	 *
  	 * If the value was already member of the set, nothing is done and 0 is
  	 * returned, otherwise the new element is added and 1 is returned. */
  	int setTypeAdd(robj *subject, sds value) {
  	    long long llval;
  	    if (subject->encoding == OBJ_ENCODING_HT) {
  	        // 若是集合編碼是字典
  
  	        dict *ht = subject->ptr;
  	        // 往字典ht中添加鍵value
  	        // 若是鍵已經存在於ht, dictAddRaw()會返回NULL
  	        // 不然會返回一個哈希節點
  	        dictEntry *de = dictAddRaw(ht,value,NULL);
  	        if (de) {
  	            // 設置節點的鍵值
  	            // 因爲是集合，因此字典節點的值設爲NULL
  	            dictSetKey(ht,de,sdsdup(value));
  	            dictSetVal(ht,de,NULL);
  	            return 1;
  	        }
  	    } else if (subject->encoding == OBJ_ENCODING_INTSET) {
  	        // 若是集合編碼是 整數集合
  
  	        // 判斷要插入的值value是否能夠轉成整數
  	        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
  	            // 若是能夠轉成整數，則能夠繼續用 整數集合 編碼
  
  	            uint8_t success = 0;
  	            // 嘗試插入的鍵值
  	            // 若是成功則 success 爲 1
  	            // 若是值已經存在，則success 爲 0
  	            subject->ptr = intsetAdd(subject->ptr,llval,&success);
  	            if (success) {
  	                // 成功，檢查集合對象中元素項的數量，若是超過限制，則轉換編碼爲字典編碼
  	                /* Convert to regular set when the intset contains
  	                 * too many entries. */
  	                if (intsetLen(subject->ptr) > server.set_max_intset_entries)
  	                    setTypeConvert(subject,OBJ_ENCODING_HT);
  	                return 1;
  	            }
  	        } else {
  	            // 若是不能夠轉成整數，則須要將編碼轉爲 字典
  	            /* Failed to get integer from object, convert to regular set. */
  	            setTypeConvert(subject,OBJ_ENCODING_HT);
  
  	            // 轉換後，字典插入鍵值
  	            /* The set *was* an intset and this value is not integer
  	             * encodable, so dictAdd should always work. */
  	            serverAssert(dictAdd(subject->ptr,sdsdup(value),NULL) == DICT_OK);
  	            return 1;
  	        }
  	    } else {
  	        serverPanic("Unknown set encoding");
  	    }
  	    return 0;
  	}

有序集合對象

有序集合對象的底層編碼是壓縮列表(ziplist)和跳躍表(skiplist)。

類型	編碼	對象
OBJ_ZSET	OBJ_ENCODING_ZIPLIST 6	壓縮列表實現的有序集合對象
OBJ_ZSET	OBJ_ENCODING_SKIPLIST 7	跳躍表實現的有序集合對象

壓縮列表實現

壓縮表內的集合元素按照分值從小到大排序，每一個集合元素使用2個壓縮表節點表示，第一個節點保存元素的成員(member)，第二個節點保存元素的分值(score)。

跳躍表實現

跳躍表的時序，是採用zset結構，zset中同時包含一個字典和一個跳躍表。

// server.h
typedef struct zset {
    dict *dict;
    zskiplist *zsl;
} zset;

結構如圖（圖來之《Redis設計與實現》）:

編碼轉換

在 Redis 4.0.11 中，當知足如下兩條件之一時，OBJ_ENCODING_ZIPLIST編碼的有序集合對象，會轉換成OBJ_ENCODING_SKIPLIST編碼的有序集合對象:

• 有序集合中保存的元素數量大於等於OBJ_ZSET_MAX_ZIPLIST_ENTRIES 128(能夠經過設置zset-max-ziplist-entries改變數值)時，OBJ_ENCODING_ZIPLIST會轉變爲OBJ_ENCODING_SKIPLIST;
• 有序集合中保存的元素，有一個或以上的長度大於等於OBJ_ZSET_MAX_ZIPLIST_VALUE 64(能夠經過設置zset-max-ziplist-value改變數值)時，OBJ_ENCODING_ZIPLIST會轉變爲OBJ_ENCODING_SKIPLIST；

若是OBJ_ENCODING_SKIPLIST中的元素知足OBJ_ZSET_MAX_ZIPLIST_ENTRIES 128和OBJ_ZSET_MAX_ZIPLIST_VALUE 64，也會編碼爲OBJ_ENCODING_ZIPLIST的。(PS: 在 Redis 4.0.11 中，只有執行ZINTERSTORE或者ZUNIONSTORE兩個命令纔會觸發，由於在代碼上看，zsetConvertToZiplistIfNeeded()函數只有在zunionInterGenericCommand()函數函數中有調用)。

部分功能代碼

• void zaddGenericCommand(client *c, int flags) ZADD和ZINCRBY命令的實現函數:

  // t_zset.c
  
  	/* This generic command implements both ZADD and ZINCRBY. */
  	void zaddGenericCommand(client *c, int flags) {
  	    ...
  
  
  	    /* Lookup the key and create the sorted set if does not exist. */
  	    檢查key是否在數據庫中
  	    zobj = lookupKeyWrite(c->db,key);
  	    if (zobj == NULL) {
  	        // Key不在數據庫中，則新建一個 有序集合對象
  
  	        if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
  	        if (server.zset_max_ziplist_entries == 0 ||
  	            server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
  	        {   // 若是設置了」zset_max_ziplist_entries「爲0，表示全部的有序集合對象 都以跳躍表實現
  	            // 檢查輸入的元素的字符長度是否超過了server.zset_max_ziplist_value，超過了則使用跳躍表編碼的有序集合
  	            zobj = createZsetObject();
  	        } else {
  	            // 沒有強制使用 跳躍表 編碼
  	            // 並且
  	            // 輸入的第一個元素字符長度小於server.zset_max_ziplist_value
  	            zobj = createZsetZiplistObject();
  	        }
  	        dbAdd(c->db,key,zobj);
  	    } else {
  	        if (zobj->type != OBJ_ZSET) {
  	            addReply(c,shared.wrongtypeerr);
  	            goto cleanup;
  	        }
  	    }
  
  	    ...
  	}

• int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) 往有序集合對象zobj中添加分值爲score的元素ele；flags是傳入的操做標識，同時也是函數結果標識；newscore若是不傳入NULL，當使用ZADD_INCR的時候，會將新值寫入newscore:

  // server.h
  	/* Input flags. */
  	#define ZADD_NONE 0
  	#define ZADD_INCR (1<<0)    /* Increment the score instead of setting it. */
  	#define ZADD_NX (1<<1)      /* Don't touch elements not already existing. */
  	#define ZADD_XX (1<<2)      /* Only touch elements already exisitng. */
  
  	/* Output flags. */
  	#define ZADD_NOP (1<<3)     /* Operation not performed because of conditionals.*/
  	#define ZADD_NAN (1<<4)     /* Only touch elements already exisitng. */
  	#define ZADD_ADDED (1<<5)   /* The element was new and was added. */
  	#define ZADD_UPDATED (1<<6) /* The element already existed, score updated. */
  
  	// t_zset.c
  
  	/* Add a new element or update the score of an existing element in a sorted
  	 * set, regardless of its encoding.
  	 *
  	 * The set of flags change the command behavior. They are passed with an integer
  	 * pointer since the function will clear the flags and populate them with
  	 * other flags to indicate different conditions.
  	 *
  	 * The input flags are the following:
  	 *
  	 * 對當前元素的分值進行增長，而不是更新操做。若是元素不存在，則以0當作以前的分值。
  	 * ZADD_INCR: Increment the current element score by 'score' instead of updating
  	 *            the current element score. If the element does not exist, we
  	 *            assume 0 as previous score.
  	 * 僅當元素不存在才執行操做
  	 * ZADD_NX:   Perform the operation only if the element does not exist.
  	 * 僅當元素存在才執行操做
  	 * ZADD_XX:   Perform the operation only if the element already exist.
  	 *
  	 * When ZADD_INCR is used, the new score of the element is stored in
  	 * '*newscore' if 'newscore' is not NULL.
  	 *
  	 * The returned flags are the following:
  	 * 返回結果的緣由:
  	 * 
  	 * 
  	 * 給定的分值並非一個數字
  	 * ZADD_NAN:     The resulting score is not a number.
  	 * 元素是新增的
  	 * ZADD_ADDED:   The element was added (not present before the call).
  	 * 元素的分值已經更新(說明元素不是新增的)
  	 * ZADD_UPDATED: The element score was updated.
  	 * 由於ZADD_NX和ZADD_XX的緣故，沒有執行操做
  	 * ZADD_NOP:     No operation was performed because of NX or XX.
  	 *
  	 * Return value:
  	 *
  	 * The function returns 1 on success, and sets the appropriate flags
  	 * ADDED or UPDATED to signal what happened during the operation (note that
  	 * none could be set if we re-added an element using the same score it used
  	 * to have, or in the case a zero increment is used).
  	 *
  	 * The function returns 0 on erorr, currently only when the increment
  	 * produces a NAN condition, or when the 'score' value is NAN since the
  	 * start.
  	 *
  	 * The commad as a side effect of adding a new element may convert the sorted
  	 * set internal encoding from ziplist to hashtable+skiplist.
  	 *
  	 * Memory managemnet of 'ele':
  	 *
  	 * The function does not take ownership of the 'ele' SDS string, but copies
  	 * it if needed. */
  	// 函數返回1表示執行成功，0 表示失敗，成功或者失敗的理由能夠參考flags的值
  	int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
  	    /* Turn options into simple to check vars. */
  	    // 若是incr不是0, 表示分值是相加操做
  	    int incr = (*flags & ZADD_INCR) != 0;
  	    // 若是nx不是0，則僅當元素不存在才執行操做
  	    int nx = (*flags & ZADD_NX) != 0;
  	    // 若是xx不是0，僅當元素存在才執行操做
  	    int xx = (*flags & ZADD_XX) != 0;
  	    *flags = 0; /* We'll return our response flags. */
  	    double curscore;
  
  	    // 判斷傳入的分值是否一個無效參數
  	    /* NaN as input is an error regardless of all the other parameters. */
  	    if (isnan(score)) {
  	        *flags = ZADD_NAN;
  	        return 0;
  	    }
  
  	    /* Update the sorted set according to its encoding. */
  	    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
  	        // 編碼爲 壓縮列表
  
  	        unsigned char *eptr;
  
  	        if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
  	            // ele元素已經存在
  
  	            /* NX? Return, same element already exists. */
  	            if (nx) {
  	                // nx(則僅當元素不存在才執行操做)不爲0
  	                // 緣由寫入flags
  	                *flags |= ZADD_NOP;
  	                return 1;
  	            }
  
  	            /* Prepare the score for the increment if needed. */
  	            if (incr) {
  	                // 分值相加操做
  
  	                score += curscore;
  	                if (isnan(score)) {
  	                    *flags |= ZADD_NAN;
  	                    return 0;
  	                }
  	                // 若是newscore不是NULL，寫入newscore，讓函數調用者能夠獲得最新的分值
  	                if (newscore) *newscore = score;
  	            }
  
  	            /* Remove and re-insert when score changed. */
  	            if (score != curscore) {
  	                // 分值變更了，將舊的節點刪除，從新插入，使得從新排序
  	                zobj->ptr = zzlDelete(zobj->ptr,eptr);
  	                zobj->ptr = zzlInsert(zobj->ptr,ele,score);
  	                *flags |= ZADD_UPDATED;
  	            }
  	            return 1;
  	        } else if (!xx) {
  	            // ele元素不存在，且xx(僅當元素存在才執行操做)爲0
  
  	            /* Optimize: check if the element is too large or the list
  	             * becomes too long *before* executing zzlInsert. */
  
  	            // 壓縮列表的整數集合插入新節點
  	            zobj->ptr = zzlInsert(zobj->ptr,ele,score);
  
  	            // 檢查整數集合的長度是否超出了「zset_max_ziplist_entries」而致使須要轉換編碼
  	            if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
  	                zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
  	            // 檢查新插入元素ele的長度是否超出了「zset_max_ziplist_value」而致使須要轉換編碼
  	            if (sdslen(ele) > server.zset_max_ziplist_value)
  	                zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
  	            if (newscore) *newscore = score;
  	            *flags |= ZADD_ADDED;
  	            return 1;
  	        } else {
  	            *flags |= ZADD_NOP;
  	            return 1;
  	        }
  	    } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
  	        // 編碼爲 跳躍表
  
  	        zset *zs = zobj->ptr;
  	        zskiplistNode *znode;
  	        dictEntry *de;
  
  	        // 在字典中查找該元素ele是否已經存在
  	        de = dictFind(zs->dict,ele);
  	        if (de != NULL) {
  	            // ele已經存在
  
  	            /* NX? Return, same element already exists. */
  	            if (nx) {
  	                // nx(則僅當元素不存在才執行操做)不爲0
  	                // 緣由寫入flags
  	                *flags |= ZADD_NOP;
  	                return 1;
  	            }
  	            curscore = *(double*)dictGetVal(de);
  
  	            /* Prepare the score for the increment if needed. */
  	            if (incr) {
  	                // 分值相加操做
  	                score += curscore;
  	                if (isnan(score)) {
  	                    *flags |= ZADD_NAN;
  	                    return 0;
  	                }
  
  	                // 若是newscore不是NULL，寫入newscore，讓函數調用者能夠獲得最新的分值
  	                if (newscore) *newscore = score;
  	            }
  
  	            /* Remove and re-insert when score changes. */
  	            if (score != curscore) {
  	                // 更新分值
  
  	                zskiplistNode *node;
  	                // 先刪除舊的節點
  	                serverAssert(zslDelete(zs->zsl,curscore,ele,&node));
  	                // 從新以新的分值插入
  	                znode = zslInsert(zs->zsl,score,node->ele);
  	                /* We reused the node->ele SDS string, free the node now
  	                 * since zslInsert created a new one. */
  	                node->ele = NULL;
  	                zslFreeNode(node);
  	                /* Note that we did not removed the original element from
  	                 * the hash table representing the sorted set, so we just
  	                 * update the score. */
  	                // 更新字典中的信息
  	                dictGetVal(de) = &znode->score; /* Update score ptr. */
  	                *flags |= ZADD_UPDATED;
  	            }
  	            return 1;
  	        } else if (!xx) {
  	            // ele存在於有序集合
  
  	            // 新建節點插入
  	            ele = sdsdup(ele);
  	            znode = zslInsert(zs->zsl,score,ele);
  	            serverAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
  	            *flags |= ZADD_ADDED;
  	            if (newscore) *newscore = score;
  	            return 1;
  	        } else {
  	            *flags |= ZADD_NOP;
  	            return 1;
  	        }
  	    } else {
  	        serverPanic("Unknown sorted set encoding");
  	    }
  	    return 0; /* Never reached. */
  	}

對象的內存回收

redisObject中的屬性refcount是記錄對象的引用計數

• 對象初始化時refcount初始化爲1；
• 當對象被一個新(函數/請求/程序)使用時，refcount增1；
• 當對象再也不被一個(函數/請求/程序)使用時，refcount減1；
• 當refcount爲0時，所佔用的內存會被釋放。

函數	做用
void decrRefCount(robj *o)	引用計數減1，若是減1後等於0，則釋放對象
void incrRefCount(robj *o)	引用計數增1
robj *resetRefCount(robj *obj)	將引用計數設爲0，但不釋放對象

使用refcount，能夠實現對象共享，Redis會在初始化服務器時，建立1萬個字符串對象，包含了 0 ~ 9999 的整數字，當須要使用到這些整數字符串對象，服務器就會使用這些共享對象，而不須要新建整數對象。