京東雲開發者| Redis數據結構(二)-List、Hash、Set及Sorted Set的結構實現

1 引言之前介紹了Redis的數據存儲及String類型的實現，接下來再來看下List、Hash、Set及Sorted Set的數據結構的實現。 2 List List類型通常被用作非同步消息隊列、文章列表查詢等；存儲有序可重覆數據或做為簡單的消息推送機制時，可以使用Redis的List類型。對於這 ...

1 引言

之前介紹了Redis的數據存儲及String類型的實現，接下來再來看下List、Hash、Set及Sorted Set的數據結構的實現。

2 List

List類型通常被用作非同步消息隊列、文章列表查詢等；存儲有序可重覆數據或做為簡單的消息推送機制時，可以使用Redis的List類型。對於這些數據的存儲通常會使用鏈表或者數組作為存儲結構。

使用數組存儲，隨機訪問節點通過索引定位時間複雜度為O(1)。但在初始化時需要分配連續的記憶體空間；在增加數據時，如果超過當前分配空間，需要將數據整體搬遷移到新數組中。
使用鏈表存儲，在進行前序遍歷或後續遍歷，當前節點中要存儲前指針和後指針，這兩個指針在分別需要8byte共16byte空間存儲，存在大量節點會因指針占用過多空間。鏈表雖然不需要連續空間存儲可以提高記憶體利用率，但頻繁的增加和刪除操作會使記憶體碎片化，影響數據讀寫速率。

如果我們能夠將鏈表和數組的特點結合起來就能夠很好處理List類型的數據存儲。

2.1 ZipList

3.2之前Redis使用的是ZipList，具體結構如下：

zlbytes: 4byte 記錄整個壓縮列表占用的記憶體位元組數：在對壓縮列表進行記憶體重分配，或者計算 zlend 的位置時使用。
zltail：4byte 記錄壓縮列表表尾節點距離壓縮列表的起始地址有多少位元組：通過這個偏移量，程式無須遍歷整個壓縮列表就可以確定表尾節點的地址。
zllen：2byte 記錄了壓縮列表包含的節點數量：當這個屬性的值小於 UINT16_MAX（65535）時，這個屬性的值就是壓縮列表包含節點的數量；當這個值等於UINT16_MAX 時，節點的真實數量需要遍歷整個壓縮列表才能計算得出。
entry X：壓縮列表包含的各個節點，節點的長度由節點保存的內容決定。包含屬性如下：
prerawlen：記錄前一個節點所占記憶體的位元組數，方便查找上一個元素地址
len：data根據len的首個byte選用不同的數據類型來存儲data
data：本元素的信息
zlend: 尾節點恆等於255

ziplist是一個連續的記憶體塊，由表頭、若幹個entry節點和壓縮列表尾部標識符zlend組成，通過一系列編碼規則，提高記憶體的利用率，使用於存儲整數和短字元串。每次增加和刪除數據時，所有數據都在同一個ziplist中都會進行搬移操作。如果將一個組數據按閾值進行拆分出多個數據，就能保證每次只操作某一個ziplist。3.2之後使用的quicklist與ziplist。

2.2 QuickList

quicklist就是維護了一種巨集觀上的雙端鏈表(類似於B樹)，鏈表的節點為對ziplist包裝的quicklistNode，每個quciklistNode都會通過前後指針相互指向，quicklist包含頭、尾quicklistNode的指針。

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* total count of all entries in all ziplists */
    unsigned long len;          /* number of quicklistNodes */
    int fill : QL_FILL_BITS;              /* fill factor for individual nodes */
    unsigned int compress : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */
   ...
} quicklist;

*head：表頭節點
*tail：表尾節點
count：節點包含entries數量
len：quicklistNode節點計數器
fill：保存ziplist的大小，配置文件設定
compress：保存壓縮程度值，配置文件設定

quicklistNode：

typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist size in bytes */
    unsigned int count : 16;     /* count of items in ziplist */
    。。。
} quicklistNode;

*prev：前置節點
*next：後置節點
*zl：不進行壓縮時指向一個ziplist結構，壓縮時指向quicklistLZF結構(具體內容請參考下方鏈接)
sz：ziplist個數
count：ziplist中包含的節點數

在redis.conf通過設置每個ziplist的最大容量，quicklist的數據壓縮範圍,提升數據存取效率,單個ziplist節點最大能存儲量，超過則進行分裂，將數據存儲在新的ziplist節點中

-5: max size: 64 Kb <— not recommended for normal workloads

-4: max size: 32 Kb <— not recommended

-3: max size: 16 Kb <— probably not recommended

-2: max size: 8 Kb <— good

-1: max size: 4 Kb <— good

List-max-ziplist-size -2
0代表所有節點，都不進行壓縮，1.代表從頭節點往後一個，尾結點往前一個不用壓縮，其它值以此類推
List-compress-depth 1

Redis 的鏈表實現的特性可以總結如下：

雙端：鏈表節點帶有prev和next指針，獲取某個節點的前置節點和後置節點的複雜度都是O(1) 。
無環：表頭節點的prev指針和表尾節點的next指針都指向NULL，對鏈表的訪問以NULL為終點。
帶表頭指針和表尾指針：通過list結構的head指針和tail指針，程式獲取鏈表的表頭節點和表尾節點的複雜度為O(1) 。
帶鏈表長度計數器：程式使用list結構的len屬性來對list持有的鏈表節點進行計數，程式獲取鏈表中節點數量的複雜度為O(1)。

3 Hash

存儲一個對象，可以直接將該對象進行序列化後使用String類型存儲，再通過反序列化獲取對象。對於只需要獲取對象的某個屬性的場景，可以將將每個屬性分別存儲；但這樣在Redis的dict中就會存在大量的key，對於鍵時效後的回收效率存在很大影響。使用Map結構就可以再dict的存儲中只存在一個key並將屬性與值再做關聯。

Redis的Hash數據結構也是使用的dict(具體實現可以查看上一篇，淺談Redis數據結構(上)-Redis數據存儲及String類型的實現)實現。當數據量比較小，或者單個元素比較小時，底層使用ziplist存儲，數據量大小和元素數量有如下配置：

ziplist元素個數超過512，將改為hashtable編碼
hash-max-ziplist-entries 512
單個元素大小超過64byte時，將改為hashtable編碼
hash-max-ziplist-value 64

4 Set

Set類型可以在對不重覆集合操作時使用，可以判斷元素是否存在於集合中。Set數據結構底層實現為value為null的dict，當數據可以使用整形表示時，Set集合將被編碼為intset結構。

typedef struct intset {
    uint32_t encoding;
    uint32_t length;
    int8_t contents[];
} intset;

整數集合是一個有序的，存儲整型數據的結構。整型集合在Redis中可以保存xxxx的整型數據，並且可以保證集合中不會出現重覆數據。

使用intset可以節省空間，會根據最大元素範圍確定所有元素類型；元素有序存儲在判斷某個元素是否存在時可以基於二分查找。但在以下任意條件不滿足時將會使用hashtable存儲數據。

元素個數大於配置的set-max-inset-entries值
元素無法用整型表示

5 Sorted Set

連續空間存儲數據，每次增加數據都會對全量數據進行搬運。對於有序鏈表查找指定元素，只能通過頭、尾節點遍歷方式進行查找，如果將每個數據增加不定層數的索引，索引之間相互關聯，尋找指定或範圍的元素時就可以通過遍歷層級的索引來確定元素所處範圍，減少空間複雜度。跳躍表是一種可以對有序鏈表進行近似二分查找的數據結構，redis 在兩個地方用到了跳躍表，一個是實現有序集合，另一個是在集群節點中用作內部數據結構。

跳躍表 ( skiplist ) 是一種有序數據結構，自動去重的集合數據類型，ZSet數據結構底層實現為字典(dict) + 跳錶(skiplist)。它通過在每個節點中維持多個指向其他節點的指針，從而達到快速訪問節點的目的。支持平均O ( logN ) 、最壞 O(N) 複雜度的節點查找，還可以通過順序性操作來批量處理節點。

數據比較少時，用ziplist編碼結構存儲，包含的元素數量比較多，又或者有序集合中元素的成員(member) 是比較長的字元串時，Redis 就會使用跳躍表來作為有序集合鍵的底層實現。

元素個數超過128，將用skiplist編碼
zset-max-ziplist-entries 128

單個元素大小超過64 byte，將用skiplist編碼
zset-max-ziplist-value 64

5.1 跳躍表

zset結構如下：

typedef struct zset {
    // 字典，存儲數據元素
    dict *dict;
    // 跳躍表，實現範圍查找
    zskiplist *zsl;
} zset;
robj *createZsetObject(void) {
    // 分配空間
    zset *zs = zmalloc(sizeof(*zs));
    robj *o;
    // dict用來查詢數據到分數的對應關係，zscore可以直接根據元素拿到分值
    zs->dict = dictCreate(&zsetDictType,NULL);
    // 創建skiplist
    zs->zsl = zslCreate();
    o = createObject(OBJ_ZSET,zs);
    o->encoding = OBJ_ENCODING_SKIPLIST;
    return o;
}

zskiplist

typedef struct zskiplist {
    // 頭、尾節點;頭節點不存儲元素，擁有最高層高
    struct zskiplistNode *header, *tail;
    unsigned long length;
    // 層級,所有節點中的最高層高
    int level;
} zskiplist;
typedef struct zskiplistNode {
    // 元素member值
    sds ele;
    // 分值
    double score;
    // 後退指針
    struct zskiplistNode *backward;
    // 節點中用 L1、L2、L3 等字樣標記節點的各個層， L1代表第一層， L2代表第二層，以此類推。
    struct zskiplistLevel {
        // 指向本層下一個節點，尾節點指向null
        struct zskiplistNode *forward;
        // *forward指向的節點與本節點之間的元素個數，span值越大，跳過的節點個數越多
        unsigned long span;
    } level[];
} zskiplistNode;

結構圖如下：

5.2 創建節點及插入流程

SkipList初始化，創建一個有最高層級的空節點：

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 分配空間
    zsl = zmalloc(sizeof(*zsl));
    // 設置起始層次
    zsl->level = 1;
    // 元素個數
    zsl->length = 0;
    // 初始化表頭，表頭不存儲元素，擁有最高的層級
    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;
    }
    // 設置表頭後退指針為NULL
    zsl->header->backward = NULL;
    // 初始表尾為NULL
    zsl->tail = NULL;
    return zsl;
}

新增元素：

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--) {
        // 存儲排位，便於更新
        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;
        }
        // 每一層的拐點
        update[i] = x;
    }
    // 隨機生成層高，以25%的概率決定是否出現下一層，越高的層出現概率越低
    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;


        // 新節點跨度為後一節點的跨度 - 兩個節點之間的跨度
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    // 新節點加入，更新頂層 span
    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
}