有序集合Sorted Set
zadd
zadd用于向集合中添加元素并且可以設(shè)置分值,比如添加三門編程語言,分值分別為1、2、3:
127.0.0.1:6379> zadd language 1 java
(integer) 1
127.0.0.1:6379> zadd language 2 c++
(integer) 1
127.0.0.1:6379> zadd language 3 python
(integer) 1
zrange
zrange根據(jù)分值區(qū)間返回符合條件的數(shù)據(jù):
127.0.0.1:6379> zrange language 1 3 withscores
1) "c++"
2) "2"
3) "python"
4) "3"
zscore
zscore根據(jù)key和元素值返回元素的分值
127.0.0.1:6379> zscore language python
"3"
Sorted Set是Redis中的一種數(shù)據(jù)結(jié)構(gòu),它可以用來存儲帶有分值的元素,并且根據(jù)分值進行排序,是一個有序的集合。
Sorted Set的結(jié)構(gòu)定義如下,它包含了一個哈希表dict和一個跳躍表zskiplist,其中哈希表可以在O(1)的時間復雜度內(nèi)進行元素查找,而跳躍表可以支持高效的范圍查詢:
typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;
跳躍表
如果一個有序集合中包含的元素數(shù)量比較多或者有序集合中的元素是比較長的字符串時,Redis就會使用跳躍表作為有序集合的底層實現(xiàn)。
跳躍表是一種多層的有序鏈表,在一個普通的有序鏈表中如果想要查找某個元素,必須遍歷鏈表,時間復雜度為O(n),那么如何提高查找效率呢,可以使用跳躍表,從列表中抽出一些元素進行分層,比如每隔一個節(jié)點就抽出一層:

此時如果需要查找元素為9的節(jié)點:
- 從第三層開始查找,元素的值為8,因為9大于8并且8之后沒有其他的節(jié)點所以接下來進入第二層
- 進入第二層,8的下一個節(jié)點為15,9小于15,所以進入第一層
- 進入第一層,獲取8的下一個節(jié)點,等于要查找的值9,查找結(jié)束
結(jié)構(gòu)定義
跳躍表的結(jié)構(gòu)定義
- header:指向跳躍表中節(jié)點的頭指針,跳躍表中的節(jié)點定義為zskiplistNode,跳躍表實際上也是一個鏈表,所以會有一個頭結(jié)點
- tail:指向跳躍表中節(jié)點的尾指針
- length:跳躍表中節(jié)點的數(shù)量
- level:跳躍表的層級
// 跳躍表
typedef struct zskiplist {
// 指向跳躍表的頭尾指針
struct zskiplistNode *header, *tail;
// 長度
unsigned long length;
// 層級
int level;
} zskiplist;
節(jié)點的結(jié)構(gòu)定義
- ele:一個sds類型的變量,存儲實際的數(shù)據(jù)
- score:存儲數(shù)據(jù)的分值,跳躍表就是按照這個分值進行排序的
- backward:一個指向前一個節(jié)點的指針,為了便于從后往前查找
- zskiplistLevel:一個層級數(shù)組,因為跳躍表可以有多層,每一層中都有一個指向當前層級中的下一個節(jié)點的指針forward和span跨度,跨度代表了當前層級里面,當前節(jié)點與下一個節(jié)點直接跨越了幾個節(jié)點
//跳躍表中的節(jié)點結(jié)構(gòu)定義
typedef struct zskiplistNode {
// 存儲的元素
sds ele;
// 分值
double score;
// 后向指針,指向當前節(jié)點的前一個節(jié)點
struct zskiplistNode *backward;
// 層級數(shù)組
struct zskiplistLevel {
// 指向當前層級中的下一個節(jié)點
struct zskiplistNode *forward;
// 跨度
unsigned long span;
} level[];
} zskiplistNode;

跳躍表的創(chuàng)建
/* 創(chuàng)建跳躍表節(jié)點*/
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
// 分配內(nèi)存
zskiplistNode *zn =
zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
zn->score = score;
zn->ele = ele;
return zn;
}
/* 創(chuàng)建跳躍表 */
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
// 跳躍表分配內(nèi)存
zsl = zmalloc(sizeof(*zsl));
// 層級初始化為1
zsl->level = 1;
// 長度為0
zsl->length = 0;
// 創(chuàng)建頭結(jié)點
zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
// 初始化每一層的頭結(jié)點
for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
zsl->header->level[j].forward = NULL;
zsl->header->level[j].span = 0;
}
// 頭結(jié)點的下一個節(jié)點指向NULL
zsl->header->backward = NULL;
// 尾節(jié)點
zsl->tail = NULL;
return zsl;
}
跳躍表層數(shù)的設(shè)置
跳躍表根據(jù)什么規(guī)則來進行層數(shù)劃分呢,有以下幾種方案:
方案一
每隔一個節(jié)點,就取出一個節(jié)點作為新的一層的節(jié)點,這樣每一層上節(jié)點的數(shù)量大約是下一層節(jié)點數(shù)的一半,此時類似于二分查找,查找復雜度為O(logn)
優(yōu)點:查找的時間復雜度降低了
缺點:由于需要維護每個層級的節(jié)點數(shù),在節(jié)點進行插入或者刪除的時候,要調(diào)整層級節(jié)點,帶來額外的開銷
方案二
新增加節(jié)點的時候,調(diào)用隨機生成層數(shù)方法,隨機生成一個當前跳躍表所需要的層數(shù),如果生成的層數(shù)等于當前層數(shù),新節(jié)點只需要加入跳躍表中即可,不需要額外的維護每一個層級的節(jié)點數(shù),Redis中就是使用的隨機生成層數(shù)的方式維護跳躍表的層級。
隨機生成層數(shù)方法:
#define ZSKIPLIST_MAXLEVEL 32 // 最大層級不超過32
#define ZSKIPLIST_P 0.25
// 隨機生成層數(shù)
int zslRandomLevel(void) {
int level = 1;
// 如果生成的隨機數(shù)的值小于ZSKIPLIST_P,層數(shù)就+1
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
// 是否超過了最大層數(shù),超過就使用最大層數(shù)
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
0xFFFF = 65535
random()&0xFFFF運算之后會生成一個0和65535之間的數(shù),ZSKIPLIST_P * 0xFFFF = 0.25 * 65535,所以random()&0xFFFF 小于 0.25 * 65535的概率為25%,也就是層數(shù)會增加1的概率不超過25%。
跳躍表增加節(jié)點
- 因為跳躍表有多層,所以需要遍歷每一層,尋找每層要插入的位置,update[i]就記錄了每一層要插入的位置
- 隨機生成跳躍表的層數(shù),如果層數(shù)有變化,則需要調(diào)整跳躍表的層高
- 創(chuàng)建節(jié)點,并將節(jié)點插入到跳躍表中
- 設(shè)置backward,新插入節(jié)點的前一個節(jié)點是update[0],如果update[0]為頭結(jié)點,當前節(jié)點的前一個節(jié)點設(shè)為null,否則backward設(shè)置為update[0]
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score));
//獲取頭結(jié)點
x = zsl->header;
/* 尋找每層要插入的位置,從高層開始向下遍歷 */
for (i = zsl->level-1; i >= 0; i--) {
// rank[i]記錄了當前層從header節(jié)點到update[i]節(jié)點所經(jīng)歷的步長
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
// 如果當前層級下一個節(jié)點不為空 并且 下一個節(jié)點的score小于要插入節(jié)點的分值 或者 下一個節(jié)點的score等于要插入節(jié)點的score并且對比兩個節(jié)點存儲的元素值之后小于0(字符串比較)
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]的值
rank[i] += x->level[i].span;
// 獲取下一個節(jié)點
x = x->level[i].forward;
}
// 記錄每層需要插入的位置
update[i] = x;
}
// 隨機生成跳躍表的層數(shù)
level = zslRandomLevel();
// 如果大于當前的層數(shù)
if (level > zsl->level) {
// 調(diào)整層數(shù)
for (i = zsl->level; i < level; i++) {
rank[i] = 0;
update[i] = zsl->header;
update[i]->level[i].span = zsl->length;
}
// 更新層數(shù)
zsl->level = level;
}
// 創(chuàng)建節(jié)點
x = zslCreateNode(level,score,ele);
// 循環(huán)每一層,添加節(jié)點
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++;
}
// 設(shè)置當前節(jié)點的前一個節(jié)點,如果update[0]為頭結(jié)點,當前節(jié)點的前一個節(jié)點設(shè)為null,否則backward設(shè)置為update[0]
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;
}
總結(jié)
- Sorted Set支持在添加元素的時候為元素添加一個分值,并根據(jù)分值排序,是一個有序的集合。
- Sorted Set在數(shù)據(jù)比較少的時候采用ziplist存儲,超過閾值后使用哈希表和跳躍表來提高查找效率,其中哈希表用于單值查詢,跳躍表用于范圍查詢。
- 跳躍表是一個多層的有序鏈表,它采用了空間換時間的方式將查找的時間復雜度降到了O(logN)。
參考
黃健宏《Redis設(shè)計與實現(xiàn)》
陳雷《Redis5設(shè)計與源碼分析》
極客時間 - Redis源碼剖析與實戰(zhàn)(蔣德鈞)
【unix21】redis源碼分析--zslRandomLevel位運算解析
【有夢想的肥宅】Redis5設(shè)計與源碼分析讀后感(三)跳躍表
Redis版本:redis-6.2.5