動態字符串SDS

字符串是Redis中最常用的一種數據結構

• Redis中的Key是字符串
• value往往是字符串或者字符串的集合

C語言字符串的缺點

Redis沒有直接用C語言中的字符串，因為C語言字符串存在一些問題：

• 獲取長度：需要\(O(n)\)遍歷數組
• 非二進制安全：以\0為結束符，則字符串中不能包含\0，不能保存像圖片、音頻、視頻文化這樣的二進制數據
• 操作不便：不可修改。進行拼接等操作時，需要考慮內存管理，存在風險

SDS

Redis構建了一種新的字符串結構，稱為簡單動態字符串（Simple Dynamic String），簡稱SDS

SDS結構

Redis 是 C 語言實現的，其中 SDS 是一個結構體，源碼如下：

struct **attribute** ((**packed**)) sdshdr8 {  
	uint8_t len; /* buf 已保存的字符串字節數，不包含結束標示 */  
	uint8_t alloc; /* buf 申請的總的字節數，不包含結束標示 */  
	unsigned char flags; /* 不同 SDS 的頭類型，用來控制 SDS 的頭大小 */  
	char buf [];  
};

• 注意：為滿足不同大小的存儲需求，還有sdshdr16、sdshdr32、sdshdr64，數字與int的bit長度相同（sdshdr5已棄用）

例如，一個包含字符串“name”的sds結構如下：

動態擴容

SDS之所以叫做動態字符串，是因為它具備動態擴容的能力。

SDS API通過alloc-len計算出可用的剩餘空間，從而判斷緩衝區大小是否支持操作正常進行。

當緩衝區大小不夠時，Redis會自動擴大SDS的空間大小

hisds hi_sdsMakeRoomFor(hisds s, size_t addlen) 
{ 
	...
	// s目前的剩餘空間已足夠，無需擴展，直接返回 
	if (avail >= addlen) return s; 
	//獲取目前s的長度 
	len = hi_sdslen(s); 
	sh = (char *)s - hi_sdsHdrSize(oldtype); 
	//擴展之後 s 至少需要的長度 
	newlen = (len + addlen); 
	//根據新長度，為s分配新空間所需要的大小 
	if (newlen < HI_SDS_MAX_PREALLOC) 
		//新長度<HI_SDS_MAX_PREALLOC 則分配所需空間*2的空間 
		newlen *= 2; 
	else 
		//否則，分配長度為目前長度+1MB 
		newlen += HI_SDS_MAX_PREALLOC; 
		... 
	}

擴容規則：

• 所需sds長度小於1MB，擴容按照翻倍擴容進行。擴容長度：兩倍的sds長度
• 所需sds長度超過1MB，會進行內存預分配。擴容長度：是sds長度+1MB
  • 預分配減少了在平凡追加時的內存分配次數（其中涉及用户態內核態的切換，性能消耗較高），優化了性能

解決了C字符串的問題

IntSet

IntSet是Redis中set集合的一種實現方式，基於整數數組來實現，並且具備長度可變、有序等特徵。

結構

typedef struct intset {
	uint32_t encoding;/*編碼方式，支持存放16位、32位、64位整數*/
	uint32_t length;/*元素個數 */
	int8_t contents[];/* 整數數組，保存集合數據*/
} intset;

其中的encoding包含三種模式，表示存儲的整數大小不同:

/* Note that these encodings are ordered, so:
*INTSET ENC INT16 < INTSET ENC INT32 < INTSET ENC INT64. */
#define INTSET_ENC_INT16(sizeof(int16_t))/* 2字節整數，範圍類似iava的short*/
#define INTSET_ENC_INT32(sizeof(int32_t))/*4字節整數，範圍類似java的int */
#define INTSET_ENC_INT64(sizeof(int64_t))/* 8字節整，範圍類似java的long */

• 由於數據元素的大小取決於其編碼方式，所以成員變量content[]相當於作為一個首元素的指針，後續遍歷也跟編碼方式匹配
• 雖然有的數據元素用不上那麼大的存儲空間，但是統一的編碼方式，有利於尋址訪問特定元素

示例

為了方便查找，Redis會將intset中所有的整數按照升序依次保存在contents數組中，結構如圖：

如上例，每個數字都在int16_t的範圍內，因此採用的編碼方式是INTSET_ENC_INT16，每部分佔用的字節大小為：

• encoding：4字節
• length：4字節
• contents：2字節*3=6字節

IntSet升級

現在假設有一個intset，元素為[5,10,20]，採用的編碼是INTSET_ENC_INT16，則每個整數佔2字節：

向其中添加一個數字：50000，超過了int16_t的範圍，intset會自動升級編碼方式到合適的大小

以此例分析升級流程

1. 升級編碼為INTSET_ENC_INT32，即每個整數佔4字節，並按照新的編碼方式及元素個數擴容數組

   • 此時所需空間：\(元素個數\times int大小=4\times4=16\)
   • 

2. 倒序依次將數組中的元素拷貝到擴容後的正確位置

   • 擴容肯定“整體後移”，如果正序進行，前面的元素擴容後會覆蓋掉後面的元素，所以倒敍進行

   • 

   • 

   • 

3. 將待添加的元素放入數組末尾

4. 最後，將intset的encoding屬性改為INTSET_ENC_INT32，將length屬性改為4

部分源碼

intset *intsetAdd(intset *is,int64_t value,uint8_t *success) {
	uint8_t valenc = _intsetValueEncoding(value);// 獲取當前值編碼
	uint32_t pos;// 要插入的位置
	if(success) *success=1;
	//判斷編碼是不是超過了當前intset的編碼
	if(valenc >intrev32ifbe(is->encoding)) {
		//超出編碼，需要升級
		return intsetUpgradeAndAdd(is,value);
	} else {
		// 在當前intset中查找值與value一樣的元素的角標pos
		if(intsetSearch(is,value,&pos)) {
			if(success) *success =0;//如果找到了，則無需插入，直接結束並返回失敗
			return is;
		//數組擴容
		is = intsetResize(is,intrev32ifbe(is->length)+1);
		//移動數組中pos之後的元素到pos+1，給新元素騰出空間
		if(pos <intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
	}
	//插入新元素
	intsetSet(is,pos,value);
	//重置元素長度
	is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
	return is;

static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
	// 獲取當前intset編碼
    uint8_t curenc = intrev32ifbe(is->encoding);
	// 獲取新編碼
    uint8_t newenc = _intsetValueEncoding(value);
	// 獲取元素個數
    int length = intrev32ifbe(is->length);
	// 判斷元素是大於還是小於0，大於0則最大，插入隊尾，小於0則最小，插入隊首
    int prepend = value < 0 ? 1 : 0;
	// 重置編碼為新編碼
    is->encoding = intrev32ifbe(newenc);
	// 重置數組大小
    is = intsetResize(is,intrev32ifbe(is->length)+1);

    /* Upgrade back-to-front so we don't overwrite values.
     * Note that the "prepend" variable is used to make sure we have an empty
     * space at either the beginning or the end of the intset. */
    while(length--)
	  _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

    /* Set the value at the beginning or the end. */
    if (prepend)
        _intsetSet(is,0,value);
    else
        _intsetSet(is,intrev32ifbe(is->length),value);
    //修改數組長度
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    return is;
}

總結

Intset可以看做是特殊的整數數組，具備一些特點：

1. Redis會確保Intset中的元素唯一、有序
2. 具備類型升級機制，可以節省內存空間
3. 底層採用二分查找方式來查詢

Dict

Redis是一個鍵值型(Key-Value Pair)的數據庫，我們可以根據鍵實現快速的增刪改查，而鍵與值的映射關係正是通過Dict來實現的

結構

Dict由三部分組成，分別是：哈希表（DictHashTable）、哈希節點（DictEntry）、字典（Dict）

typedef struct dictht {
	// entry數組
	// 數組中保存的是指向entry的指針
	dictEntry **table;
	// 哈希表大小(總是2的n次方)
	unsigned long size;
	// 哈希表大小的掩碼，總等於size -1
	unsigned long sizemask;
	// entry個數
	unsigned long used;
} dictht;

typedef struct dictEntry {
	void *key;// 鍵
	union {
		void *val:
		uint64 t u64;
		int64_t s64;
		double d;
	} v;//值
	//下一個Entry的指針
	struct dictEntry *next;
} dictEntry;

• *next：哈希衝突時使用，拉鍊法

typedef struct dict {
	dictType *type;//dict類型，內置不同的hash函數
	void *privdata;//私有數據，在做特殊hash運算時用
	dictht ht[2];//一個Dict包含兩個哈希表，其中一個是當前數據，另一個一般是空，rehash時使用
	long rehashidx; //rehash的進度，-1表示未進行
	int16 t pauserehash:// rehash是否暫停，1則暫停，0則繼續
} dict;

添加元素

當我們向Dict添加鍵值對時，Redis首先根據key計算出hash值(h)，然後利用h & sizemask（實際上等效於取餘，位運算的性能好）來計算元素應該存儲到數組中的哪個索引位置。

• 假設k1的哈希值h=1，則1&3=1，因此k1=v1要存儲到數組角標1位置

• 如果發生哈希衝突，採用頭插法（便於操作，如果尾插則需遍歷到結尾再插入）

Dict的擴容

Dict中的HashTable就是數組結合單向鏈表的實現，當集合中元素較多時，必然導致哈希衝突增多，鏈表過長，則查詢效率會大大降低
Dict在每次新增鍵值對時都會檢查負載因子(LoadFactor=used/size)，滿足以下兩種情況時會觸發哈希表擴容:

• 哈希表的 LoadFactor>=1，並且服務器沒有執行 BGSAVE 或者 BGREWRITEAOF 等後台進程;
  • 這些後台進程對CPU使用較多，且涉及IO操作，性能開銷大
• 哈希表的 LoadFactor>5
  • 過大了，嚴重影響了性能，於是需要馬上擴容

static int _dictExpandIfNeeded(dict *d)
{
    //如果正在rehash，則返回ok
    if (dictIsRehashing(d)) return DICT_OK;
    //如果哈希表為空，則初始化哈希表為默認大小:4
    if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);
    //當負載因子(used/size)達到1以上，並且當前沒有進行bgrewrite等子進程操作
	//或者負載因子超過5，則進行 dictExpand ，也就是擴容
    if (d->ht[0].used >= d->ht[0].size &&
        (dict_can_resize ||
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio) &&
        dictTypeExpandAllowed(d))
    {
    	//擴容大小為used + 1，底層會對擴容大小做判斷，實際上找的是第一個大於等於 used+1 的 2^n
        return dictExpand(d, d->ht[0].used + 1);
    }
    return DICT_OK;
}

Dict的收縮

Dict除了擴容以外，每次刪除元素時，也會對負載因子做檢查，當LoadFactor<0.1時，會做哈希表收縮:

//t_hash.c # hashTypeDeleted()
...
if(dictDelete((dict*)o->ptr,field)==C_0K) {
	deleted-1:
	//制除成功後，檢查是否需要重置Dict大小，如果需要則調用dictResize重置
	/* Always check if the dictionary needs a resize after a delete. */
	if(htNeedsResize(o->ptr)) dictResize(o->ptr);
}
...

//server.c 文件
int htNeedsResize(dict *dict){
	long long size, used;
	// 哈希表大小
	size = dictSlots(dict);
	//entry數量
	used = dictsize(dict);
	//size>4(哈希表初識大小)並且 負載因子低於0.1
	return(size>DICT HT INITIAL SIZE &&(used*10@/size < HASHTABLE_MIN_FILL));
}

int dictResize(dict *d) {
	unsigned long minimal;
	// 如果正在做bgsave或bgrewriteof或rehash，則返回錯誤
	if(!dict_can_resize||dictIsRehashing(d))
		return DICT_ERR;
	//獲取used，也就是entry個數
	minimal = d->ht[0].used;
	//如果used小於4，則重置為4
	if(minimal < DICTHT_INITIAL_SIZE)
		minimal = DICT_HT_INITIAL_SIZE;
	//重置大小為minimal，其實是第一個大於等於minimal的2^n
	return dictExpand(d,minimal);
}

Dict的rehash

rehash

不管是擴容還是收縮，必定會創建新的哈希表，導致哈希表的size和sizemask變化，而key的查詢與sizemask有關。
因此必須對哈希表中的每一個key重新計算索引，插入新的哈希表，這個過程稱為rehash。
過程是這樣的:

1. 計算新hash表的realeSize，值取決於當前要做的是擴容還是收縮:
   • 如果是擴容，則新size為第一個大於等於dict.ht[0].used+1的\(2^n\)
   • 如果是收縮，則新size為第一個大於等於dict.ht[0].used的\(2^n\)(不得小於4)
2. 按照新的realeSize申請內存空間，創建dictht，並賦值給dict.ht[1]
3. 設置dict.rehashidx=0，標示開始rehash
4. 將dict.ht[0]中的每一個dictEntry都rehash到dict.ht[1]
5. 將dict.ht[1]賦值給dict.ht[0]，給dict.ht[1]初始化為空哈希表，釋放原來的dict.ht[0]的內存

下面看一個具體過程

1. 初始狀態

2. 新增一個元素，現在需要擴容

3. 申請空間並分配給ht[1]，同時修改其頭部信息，將dict的rehashidx置0

4. 將元素映射到ht[1]中

5. 然ht[0]指向ht[1]的dictEntry數組，ht[1]置空，修改頭部信息及dict的rehashidx字段

漸進式

Dict的rehash並不是一次性完成的。試想一下，如果Dict中包含數百萬的entry，要在一次rehash完成，極有可能導致主線程阻塞。
因此Dict的rehash是分多次、漸進式的完成，因此稱為漸進式rehash。流程如下:

1. 計算新hash表的size，值取決於當前要做的是擴容還是收縮:
   • 如果是擴容，則新size為第一個大於等於dict.ht[0].used+1的2”
   • 如果是收縮，則新size為第一個大於等於dict.ht[0].used的2”(不得小於4)
2. 按照新的size申請內存空間，創建dictht，並賦值給dict.ht[1]
3. 設置dict.rehashidx=0，標示開始rehash
4. 每次執行新增、查詢、修改、刪除操作時，都檢查一下dict.rehashidx是否大於-1，如果是則將dict.ht[0].table[rehashidx]的entry鏈表rehash到dict.ht[1]，並且將rehashidx++。直到dict.ht[0]的所有數據都rehash到dict.ht[1]
5. 將dict.ht[1]賦值給dict.ht[0]，給dict.ht[1]初始化為空哈希表，釋放原來的dict.ht[0]的內存
6. 將rehashidx賦值為-1，代表rehash結束
7. 在rehash過程中，新增操作，則直接寫入ht[1]，查詢、修改和刪除則會在dict.ht[0]和dict.ht[1]依次查找並執行。這樣可以確保ht[0]的數據只減不增，隨着rehash最終為空

總結

Dict的結構:

• 類似java的HashTable，底層是數組加鏈表來解決哈希衝突
• Dict包含兩個哈希表，ht[0]平常用，ht[1]用來rehash

Dict的伸縮:

• 當LoadFactor大於5或者LoadFactor大於1並且沒有子進程任務時，Dict擴容
• 當LoadFactor小於0.1時，Dict收縮
• 擴容大小為第一個大於等於used+1的\(2^n\)
• 收縮大小為第一個大於等於used 的\(2^n\)
• Dict採用漸進式rehash，每次訪問Dict時執行一次rehash
• rehash時ht[0]只減不增，新增操作只在ht[1]執行，其它操作在兩個哈希表

ZipList

ZipList是一種特殊的“雙端鏈表”，由一系列特殊編碼的連續內存塊組成。可以在任意一端進行壓入/彈出操作，並且該操作的時間複雜度為\(O(1)\)

ZipList結構

ZipListEntry

ZipList中的Entry並不像普通鏈表那樣記錄前後節點的指針，因為記錄兩個指針要佔用16個字節，浪費內存。而是採用了下面的結構:

• previous entry length：前一節點的長度，佔1個或5個字節。
  • 如果前一節點的長度小於254字節，則採用1個字節來保存這個長度值
  • 如果前一節點的長度大於254字節，則採用5個字節來保存這個長度值，第一個字節為0xfe，後四個字節才是真實長度數據
• encoding：編碼屬性，記錄content的數據類型(字符串還是整數)以及長度，佔用1個、2個或5個字節
• contents：負責保存節點的數據，可以是字符串或整數

注意：ZipList中所有存儲長度的數值均採用小端字節序，即地位字節在前，高位字節在後。
例如：數值0x1234，採用小端字節序實際存儲值為：0x3412

Encoding編碼

ZipListEntry中的encoding編碼分為字符串和整數兩種：

字符串

字符串：如果encoding是以“00”、“01”或者“10”開頭，則證明content是字符串

例如：
一個ZipList保存字符串：“ab”、“bc”

整數

整數：如果encoding是以“11”開始，則證明content是整數，且encoding固定只佔用1個字節

例如，一個ZipList保存兩個整數值：“2”、“5”

ZipList的連鎖更新問題

ZipList的每個Entry都包含previous_entry_length來記錄上一個節點的大小，長度是1個或5個字節：

• 如果前一節點的長度小於254字節，則採用1個字節來保存這個長度值
• 如果前一節點的長度大於等於254字節，則採用5個字節來保存這個長度值，第一個字節為0xfe，後四個字節才是真實長度數據

現在，假設我們有N個連續的、長度為250~253字節之間的entry，因此entry的previous_entry_length屬性用1個字節即可表示。

1. 初始狀態

2. 插入一條長度為254字節的數據

   • 新插入的元素導致後繼節點的pre_entry_len存儲變化，進而導致後繼節點的長度變化，並引發了其後繼多個節點的連鎖反應

ZipList這種特殊情況下產生的連續多次空間擴展操作稱之為連鎖更新（Cascade Update）。新增、刪除都可能導致連鎖更新的發生。

總結

ZipList特性：

1. 壓縮列表的可以看做一種連續內存空間的"雙向鏈表"
2. 列表的節點之間不是通過指針連接，而是記錄上一節點和本節點長度來尋址，內存佔用較低
3. 如果列表數據過多，導致鏈表過長，可能影響查詢性能
4. 增或刪較大數據時有可能發生連續更新問題

QuickList

問題1：ZipList雖然節省內存，但申請內存必須是連續空間，如果內存佔用較多，申請內存效率很低。怎麼辦？

• 為了緩解這個問題，我們必須限制ZipList的長度和entry大小。
  問題2：但是我們要存儲大量數據，超出了ZipList最佳的上限該怎麼辦？
• 我們可以創建多個ZipList來分片存儲數據。
  問題3：數據拆分後比較分散，不方便管理和查找，這多個ZipList如何建立聯繫？
• Redis在3.2版本引入了新的數據結構QuickList，它是一個雙端鏈表，只不過鏈表中的每個節點都是一個ZipList。
  

結構

以下是QuickList的和QuickListNode的結構源碼：

typedef struct quicklist {
    // 頭節點指針
    quicklistNode *head; 
    // 尾節點指針
    quicklistNode *tail; 
    // 所有ziplist的entry的數量
    unsigned long count;    
    // ziplists總數量
    unsigned long len;
    // ziplist的entry上限，默認值 -2 
    int fill : QL_FILL_BITS;     
    // 首尾不壓縮的節點數量
    unsigned int compress : QL_COMP_BITS;
    // 內存重分配時的書籤數量及數組，一般用不到
    unsigned int bookmark_count: QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

typedef struct quicklistNode {
    // 前一個節點指針
    struct quicklistNode *prev;
    // 下一個節點指針
    struct quicklistNode *next;
    // 當前節點的ZipList指針
    unsigned char *zl;
    // 當前節點的ZipList的字節大小
    unsigned int sz;
    // 當前節點的ZipList的entry個數
    unsigned int count : 16;  
    // 編碼方式：1，ZipList; 2，lzf壓縮模式
    unsigned int encoding : 2;
    // 數據容器類型（預留）：1，其它；2，ZipList
    unsigned int container : 2;
    // 是否被解壓縮。1：則説明被解壓了，將來要重新壓縮
    unsigned int recompress : 1;
    unsigned int attempted_compress : 1; //測試用
    unsigned int extra : 10; /*預留字段*/
} quicklistNode;

配置

ZipList的entry數量

為了避免QuickList中的每個ZipList中entry過多，Redis提供了一個配置項：list-max-ziplist-size來限制。

• 如果值為正，則代表ZipList的允許的entry個數的最大值
• 如果值為負，則代表ZipList的最大內存大小，分5種情況：
  1. -1：每個ZipList的內存佔用不能超過4kb
  2. -2：每個ZipList的內存佔用不能超過8kb
  3. -3：每個ZipList的內存佔用不能超過16kb
  4. -4：每個ZipList的內存佔用不能超過32kb
  5. -5：每個ZipList的內存佔用不能超過64kb
     其默認值為 -2：
     

除了控制ZipList的大小，QuickList還可以對節點的ZipList做壓縮。通過配置項list-compress-depth來控制。因為鏈表一般都是從首尾訪問較多，所以首尾是不壓縮的。這個參數是控制首尾不壓縮的節點個數：

• 0：特殊值，代表不壓縮
• 1：標示QuickList的首尾各有1個節點不壓縮，中間節點壓縮
• 2：標示QuickList的首尾各有2個節點不壓縮，中間節點壓縮
• 以此類推
  默認值：
  

總結

QuickList的特點：

• 是一個節點為ZipList的雙端鏈表
• 節點採用ZipList，解決了傳統鏈表的內存佔用問題
• 控制了ZipList大小，解決連續內存空間申請效率問題
• 中間節點可以壓縮，進一步節省了內存

SkipList

SkipList（跳錶） 首先是鏈表，但與傳統鏈表相比有幾點差異

• 元素按照升序排列存儲
• 節點可能包含多個指針，指針跨度不同

結構

// t_zset.c
typedef struct zskiplist {
    // 頭尾節點指針
    struct zskiplistNode *header, *tail;
    // 節點數量
    unsigned long length;
    // 最大的索引層級，默認是1
    int level;
} zskiplist;

// t_zset.c
typedef struct zskiplistNode {
    sds ele; // 節點存儲的值
    double score;// 節點分數，排序、查找用
    struct zskiplistNode *backward; // 前一個節點指針
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 下一個節點指針
        unsigned long span; // 索引跨度
    } level[]; // 多級索引數組
} zskiplistNode;

總結

SkipList的特點：

• 跳躍表是一個雙向鏈表，每個節點都包含score和ele值
• 節點按照score值排序，score值一樣則按照ele字典排序
• 每個節點都可以包含多層指針，層數是1到32之間的隨機數（有相關算法進行設計）
• 不同層指針到下一個節點的跨度不同，層級越高，跨度越大
• 增刪改查效率與紅黑樹基本一致，實現卻更簡單

RedisObject

Redis中的任意數據類型的鍵和值都會被封裝成一個RedisObject，也叫做Redis對象，源碼如下：

redisObject頭部就需要佔16字節

注意：如果有大量String對象，每一個對象都有16字節的頭部信息，就會造成很大的內存開銷，所以存儲大量數據時，建議不要用String類型，而用集合類型

Redis的編碼方式

Redis中會根據存儲的數據類型不同，選擇不同的編碼方式，共包含11種不同類型：

五種數據類型的編碼方式

五種數據結構

String

String是Redis中最常見的數據存儲類型：

• 其基本編碼方式是RAW，基於簡單動態字符串（SDS）實現，存儲上限為512mb。

• 如果存儲的SDS長度小於44字節，則會採用EMBSTR編碼，此時object head與SDS是一段連續空間。申請內存時只需要調用一次內存分配函數，效率更高。

• 如果存儲的字符串是整數值，並且大小在LONG_MAX範圍內，則會採用INT編碼：直接將數據保存在RedisObject的ptr指針位置（剛好8字節），不再需要SDS了。

為什麼以44字節為界?

當44字節時，SDS結構如圖

頭佔3字節、尾佔1個字節，內容佔44字節
所以SDS共佔48字節
加上RedisObject的16字節，整體共佔64字節

而Redis的內存分配算法會以\(2^n\)進行分配，64恰好是一個分片大小，不會產生內存碎片

總結

• 使用字符串時，儘可能保證其大小不超過44字節
• 能使用整型保存就使用整型保存

List

Redis的List支持從首位對元素進行操作，哪個數據結構比較合適呢

• LinkedList ：普通鏈表，可以從雙端訪問，內存佔用較高，內存碎片較多
• ZipList ：壓縮列表，可以從雙端訪問，內存佔用低，存儲上限低
• QuickList：LinkedList + ZipList，可以從雙端訪問，內存佔用較低，包含多個ZipList，存儲上限高

在3.2版本之前，Redis採用ZipList和LinkedList來實現List，當元素數量小於512並且元素大小小於64字節時採用ZipList編碼，超過則採用LinkedList編碼。
在3.2版本之後，Redis統一採用QuickList來實現List：

void pushGenericCommand(client *c, int where, int xx) {
    int j;
    // 嘗試找到KEY對應的list
    robj *lobj = lookupKeyWrite(c->db, c->argv[1]);
    // 檢查類型是否正確
    if (checkType(c,lobj,OBJ_LIST)) return;
    // 檢查是否為空
    if (!lobj) {
        if (xx) {
            addReply(c, shared.czero);
            return;
        }
        // 為空，則創建新的QuickList
        lobj = createQuicklistObject();
        quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
                            server.list_compress_depth);
        dbAdd(c->db,c->argv[1],lobj);
    }
    // 略 ...
}

robj *createQuicklistObject(void) {
    // 申請內存並初始化QuickList
    quicklist *l = quicklistCreate();
    // 創建RedisObject，type為OBJ_LIST
    // ptr指向 QuickList
    robj *o = createObject(OBJ_LIST,l);
    // 設置編碼為 QuickList
    o->encoding = OBJ_ENCODING_QUICKLIST;
    return o;
}

內存結構

Set

Set是Redis中的單列集合，滿足下列特點：

• 不保證有序性
• 保證元素唯一（可以判斷元素是否存在）
• 求交集、並集、差集

可以看出，Set對查詢元素的效率要求非常高，思考一下，什麼樣的數據結構可以滿足？

• HashTable，也就是Redis中的Dict，不過Dict是雙列集合（可以存鍵、值對）

Set是Redis中的集合，不一定確保元素有序，可以滿足元素唯一、查詢效率要求極高。

• 為了查詢效率和唯一性，set採用HT編碼（Dict）。Dict中的key用來存儲元素，value統一為null。
• 當存儲的所有數據都是整數，並且元素數量不超過set-max-intset-entries時，Set會採用IntSet編碼，以節省內存
  • 二分查找保證了查詢以及判斷是否存在的效率

創建

robj *setTypeCreate(sds value) {
    // 判斷value是否是數值類型 long long 
    if (isSdsRepresentableAsLongLong(value,NULL) == C_OK)
        // 如果是數值類型，則採用IntSet編碼
        return createIntsetObject();
    // 否則採用默認編碼，也就是HT
    return createSetObject();
}

robj *createIntsetObject(void) {
    // 初始化INTSET並申請內存空間
    intset *is = intsetNew();
    // 創建RedisObject
    robj *o = createObject(OBJ_SET,is);
    // 指定編碼為INTSET
    o->encoding = OBJ_ENCODING_INTSET;
    return o;
}

robj *createSetObject(void) {
    // 初始化Dict類型，並申請內存
    dict *d = dictCreate(&setDictType,NULL);
    // 創建RedisObject
    robj *o = createObject(OBJ_SET,d);
    // 設置encoding為HT
    o->encoding = OBJ_ENCODING_HT;
    return o;
}

類型轉換

int setTypeAdd(robj *subject, sds value) {
    long long llval;
    if (subject->encoding == OBJ_ENCODING_HT) {
        dict *ht = subject->ptr;
        dictEntry *de = dictAddRaw(ht,value,NULL);
        if (de) {
            dictSetKey(ht,de,sdsdup(value));
            dictSetVal(ht,de,NULL);
            return 1;
        }
    } else if (subject->encoding == OBJ_ENCODING_INTSET) {
        if (isSdsRepresentableAsLongLong(value,&llval) == C_OK) {
            uint8_t success = 0;
            subject->ptr = intsetAdd(subject->ptr,llval,&success);
            if (success) {
                /* Convert to regular set when the intset contains
                 * too many entries. */
                size_t max_entries = server.set_max_intset_entries;
                /* limit to 1G entries due to intset internals. */
                if (max_entries >= 1<<30) max_entries = 1<<30;
                if (intsetLen(subject->ptr) > max_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;
}

兩種發生類型轉換的情況

• 原本是intset實現，但是添加了不是int類型的元素
• 元素都是int，但是元素個數超出了server.set_max_intset_entries的值時
  • server.set_max_intset_entries默認值為512，可進行設置，一般不建議設置過大，否則會影響查詢效率

內存及類型轉換圖

Zset

ZSet也就是SortedSet，其中每一個元素都需要指定一個score值和member值：

• 可以根據score值排序後
• member必須唯一
• 可以根據member查詢分數

因此，zset底層數據結構必須滿足鍵值存儲、鍵必須唯一、可排序這幾個需求。之前學習的哪種編碼結構可以滿足？

• SkipList：
  • 優點：可以排序，並且可以同時存儲score和ele值（member）
  • 缺點：鍵值唯一較難處理；根據member查score需要遍歷整個鏈表
• HT（Dict）：
  • 優點：可以鍵值存儲，並且可以根據key找value
  • 缺點：不能做排序

結構

Zset：“小孩子才做選擇，我全都要！”

Zset結構

// zset結構
typedef struct zset {
    // Dict指針
    dict *dict;
    // SkipList指針
    zskiplist *zsl;
} zset;

創建Zset

robj *createZsetObject(void) {
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;
    // 創建Dict
    zs->dict = dictCreate(&zsetDictType,NULL);
    // 創建SkipList
    zs->zsl = zslCreate(); 
    o = createObject(OBJ_ZSET,zs);
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

內存圖

由圖可以直觀看出，內存佔用太高

• 數據重複存儲、存在大量指針……

另一種實現

當元素數量不多時，HT和SkipList的優勢不明顯，而且更耗內存。因此zset還會採用ZipList結構來節省內存，不過需要同時滿足兩個條件：

1. 元素數量小於zset_max_ziplist_entries，默認值128
2. 每個元素都小於zset_max_ziplist_value字節，默認值64

// zadd添加元素時，先根據key找到zset，不存在則創建新的zset
zobj = lookupKeyWrite(c->db,key);
if (checkType(c,zobj,OBJ_ZSET)) goto cleanup;
// 判斷是否存在
if (zobj == NULL) { // zset不存在
    if (server.zset_max_ziplist_entries == 0 ||
        server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
    { // zset_max_ziplist_entries設置為0就是禁用了ZipList，
        // 或者value大小超過了zset_max_ziplist_value，採用HT + SkipList
        zobj = createZsetObject();
    } else { // 否則，採用 ZipList
        zobj = createZsetZiplistObject();
    }
    dbAdd(c->db,key,zobj); 
}
// ....
zsetAdd(zobj, score, ele, flags, &retflags, &newscore);

robj *createZsetObject(void) {
    // 申請內存
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;
    // 創建Dict
    zs->dict = dictCreate(&zsetDictType,NULL);
    // 創建SkipList
    zs->zsl = zslCreate();
    o = createObject(OBJ_ZSET,zs);
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

robj *createZsetZiplistObject(void) {
    // 創建ZipList
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_ZSET,zl);
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

編碼轉換

int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
    /* 判斷編碼方式*/
    if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {// 是ZipList編碼
        unsigned char *eptr;
        // 判斷當前元素是否已經存在，已經存在則更新score即可        if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
            //...略
            return 1;
        } else if (!xx) {
            // 元素不存在，需要新增，則判斷ziplist長度有沒有超、元素的大小有沒有超
            if (zzlLength(zobj->ptr)+1 > server.zset_max_ziplist_entries
 		|| sdslen(ele) > server.zset_max_ziplist_value 
 		|| !ziplistSafeToAdd(zobj->ptr, sdslen(ele)))
            { // 如果超出，則需要轉為SkipList編碼
                zsetConvert(zobj,OBJ_ENCODING_SKIPLIST);
            } else {
                zobj->ptr = zzlInsert(zobj->ptr,ele,score);
                if (newscore) *newscore = score;
                *out_flags |= ZADD_OUT_ADDED;
                return 1;
            }
        } else {
            *out_flags |= ZADD_OUT_NOP;
            return 1;
        }
    }    // 本身就是SKIPLIST編碼，無需轉換
    if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
       // ...略
    } else {
        serverPanic("Unknown sorted set encoding");
    }
    return 0; /* Never reached. */
}

注意到轉換條件：

1. 壓縮列表中的元素數量加 1 後超過了配置項 zset_max_ziplist_entries 的值時
2. 當要插入的元素（ele）的長度超過了配置項 zset_max_ziplist_value 的值時
3. ziplistSafeToAdd 函數用於檢查是否有足夠的內存空間來將新元素添加到壓縮列表中。若該函數返回 false，也就是沒有足夠的內存空間，就會觸發轉換。（ZipList需要的是連續的內存空間）

ZipList如何滿足Zset需求

Ziplist本身沒有排序功能，而且沒有鍵值對的概念，因此需要有zset通過編碼實現：

• ZipList是連續內存，因此score和element是緊挨在一起的兩個entry， element在前，score在後
• score越小越接近隊首，score越大越接近隊尾，按照score值升序排列
  

Hash

Hash結構與Redis中的Zset非常類似：

• 都是鍵值存儲
• 都需求根據鍵獲取值
• 鍵必須唯一
  區別如下：
• zset的鍵是member，值是score；hash的鍵和值都是任意值
• zset要根據score排序；hash則無需排序

結構

因此，Hash底層採用的編碼與Zset也基本一致，只需要把排序有關的SkipList去掉即可

• Hash結構默認採用ZipList編碼，用以節省內存。 ZipList中相鄰的兩個entry 分別保存field和value
• 當數據量較大時，Hash結構會轉為HT編碼，也就是Dict，觸發條件有兩個：
  1. ZipList中的元素數量超過了hash-max-ziplist-entries（默認512）
  2. ZipList中的任意entry大小超過了hash-max-ziplist-value（默認64字節）

ZipList形式內存圖：

Dict形式內存圖：

源碼

當你執行一條hset命令時……

void hsetCommand(client *c) {// hset user1 name Jack age 21
    int i, created = 0;
    robj *o; // 略 ...
     // 判斷hash的key是否存在，不存在則創建一個新的，默認採用ZipList編碼
    if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
    // 判斷是否需要把ZipList轉為Dict
    hashTypeTryConversion(o,c->argv,2,c->argc-1);
    // 循環遍歷每一對field和value，並執行hset命令
    for (i = 2; i < c->argc; i += 2)
        created += !hashTypeSet(o,c->argv[i]->ptr,c->argv[i+1]->ptr,HASH_SET_COPY);    // 略 ...
}

判斷是需要寫入還是需要創建hash

robj *hashTypeLookupWriteOrCreate(client *c, robj *key) {
    // 查找key
    robj *o = lookupKeyWrite(c->db,key);
    if (checkType(c,o,OBJ_HASH)) return NULL;
    // 不存在，則創建新的
    if (o == NULL) {
        o = createHashObject();
        dbAdd(c->db,key,o);
    }
    return o;
}

創建hash

robj *createHashObject(void) {
    // 默認採用ZipList編碼，申請ZipList內存空間
    unsigned char *zl = ziplistNew();
    robj *o = createObject(OBJ_HASH, zl);
    // 設置編碼
    o->encoding = OBJ_ENCODING_ZIPLIST;
    return o;
}

判斷hash是否需要進行編碼轉換
hashTypeTryConversion(o,c->argv,2,c->argc-1);

• 以hset user1 name Jack age 21為例，下標為2的參數開始才是存的數據，需要被判斷

void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {
    int i;
    size_t sum = 0;
    // 本來就不是ZipList編碼，什麼都不用做了
    if (o->encoding != OBJ_ENCODING_ZIPLIST) return;
    // 依次遍歷命令中的field、value參數
    for (i = start; i <= end; i++) {
        if (!sdsEncodedObject(argv[i]))
            continue;
        size_t len = sdslen(argv[i]->ptr);
        // 如果field或value超過hash_max_ziplist_value，則轉為HT
        if (len > server.hash_max_ziplist_value) {
            hashTypeConvert(o, OBJ_ENCODING_HT);
            return;
        }
        sum += len;
    }// ziplist大小超過1G，也轉為HT
    if (!ziplistSafeToAdd(o->ptr, sum))
        hashTypeConvert(o, OBJ_ENCODING_HT);
}

執行set過程

int hashTypeSet(robj *o, sds field, sds value, int flags) {
    int update = 0;
    // 判斷是否為ZipList編碼
    if (o->encoding == OBJ_ENCODING_ZIPLIST) {
        unsigned char *zl, *fptr, *vptr;
        zl = o->ptr;
        // 查詢head指針
        fptr = ziplistIndex(zl, ZIPLIST_HEAD);
        if (fptr != NULL) { // head不為空，説明ZipList不為空，開始查找key
            fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
            if (fptr != NULL) {// 判斷是否存在，如果已經存在則更新
                update = 1;
                zl = ziplistReplace(zl, vptr, (unsigned char*)value,
                        sdslen(value));
            }
        }
        // 不存在，則直接push
        if (!update) { // 依次push新的field和value到ZipList的尾部
            zl = ziplistPush(zl, (unsigned char*)field, sdslen(field),
                    ZIPLIST_TAIL);
            zl = ziplistPush(zl, (unsigned char*)value, sdslen(value),
                    ZIPLIST_TAIL);
        }
        o->ptr = zl;
        /* 插入了新元素，檢查list長度是否超出，超出則轉為HT */
        if (hashTypeLength(o) > server.hash_max_ziplist_entries)
            hashTypeConvert(o, OBJ_ENCODING_HT);
    } else if (o->encoding == OBJ_ENCODING_HT) {
        // HT編碼，直接插入或覆蓋
    } else {
        serverPanic("Unknown hash encoding");
    }
    return update;
}