Redis 原始碼簡潔剖析 04 - Sorted Set 有序集合
阿新 • • 發佈:2022-01-29
Sorted Set 是什麼
有序集合(Sorted Set)
是 Redis 中一種重要的資料型別,它本身是集合型別,同時也可以支援集合中的元素帶有權重,並按權重排序。
- ZRANGEBYSCORE:按照元素權重返回一個範圍內的元素
- ZSCORE:返回某個元素的權重值
Sorted Set 命令及實現方法
Sorted Set 資料結構
- 結構定義:
server.h
- 實現:
t_zset.c
結構定義是 zset,裡面包含雜湊表 dict
和跳錶 zsl
。zset 充分利用了:
- 雜湊表的高效單點查詢特性(ZSCORE)
- 跳錶的高效範圍查詢(ZRANGEBYSCORE)
typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;
跳錶(skiplist)
多層的有序連結串列。下面展示的是 3 層的跳錶,頭節點是一個 level 陣列,作為 level0~level2 的頭指標。
跳錶節點的結構定義
typedef struct zskiplistNode { // sorted set 中的元素 sds ele; // 元素權重 double score; // 後向指標(為了便於從跳錶的尾節點倒序查詢) struct zskiplistNode *backward; // 節點的 level 陣列 struct zskiplistLevel { // 每層上的前向指標 struct zskiplistNode *forward; // 跨度,記錄節點在某一層 *forward 指標和該節點,跨越了 level0 上的幾個節點 unsigned long span; } level[]; } zskiplistNode;
跳錶的定義
typedef struct zskiplist {
// 頭節點和尾節點
struct zskiplistNode *header, *tail;
unsigned long length;
int level;
} zskiplist;
跳錶節點查詢
在查詢某個節點時,跳錶會從頭節點的最高層開始,查詢下一個節點:
- 訪問下一個節點
- 當前節點的元素權重 < 要查詢的權重
- 當前節點的元素權重 = 要查詢的權重,且節點資料<要查詢的資料
- 訪問當前節點 level 陣列的下一層指標
- 當前節點的元素權重 > 要查詢的權重
//獲取跳錶的表頭
x = zsl->header;
//從最大層數開始逐一遍歷
for (i = zsl->level-1; i >= 0; i--) {
...
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))) {
...
x = x->level[i].forward;
}
...
}
層數設定
幾種方法:
- 每層的節點數約是下一層節點數的一半。
- 好處:查詢時類似於二分查詢,查詢複雜度可以減低到 O(logN)
- 壞處:每次插入/刪除節點,都要調整後續節點層數,帶來額外開銷
隨機生成每個節點的層數
。Redis 跳錶採用了這種方法。
Redis 中,跳錶節點層數是由 zslRandomLevel 函式決定。
int zslRandomLevel(void) {
int level = 1;
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
其中每層增加的概率是 0.25,最大層數是 32。
#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */
#define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */
跳錶插入節點 zslInsert
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;
// 從最高層的 level 開始找
for (i = zsl->level-1; i >= 0; i--) {
// 每層待插入的位置
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
// forward.score < 待插入 score || (forward.score < 待插入 score && forward.ele < ele)
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))) {
// 在同一層 level 找下一個節點
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
// 隨機層數
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);
// 設定新建節點的 level 陣列
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 */
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++;
}
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;
}
跳錶刪除節點 zslDelete
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
int i;
x = zsl->header;
// 找到待刪除的節點
for (i = zsl->level-1; i >= 0; i--) {
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)))
{
x = x->level[i].forward;
}
update[i] = x;
}
x = x->level[0].forward;
// 判斷節點的 score 和 ele 是否符合條件
if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
// 刪除該節點
zslDeleteNode(zsl, x, update);
if (!node)
// 釋放記憶體
zslFreeNode(x);
else
*node = x;
return 1;
}
return 0; /* not found */
}
Sorted Set 基本操作
首先看下如何建立跳錶,程式碼在 object.c 中,可以看到會呼叫 dictCreate 函式建立雜湊表,之後呼叫 zslCreate 函式建立跳錶。
robj *createZsetObject(void) {
zset *zs = zmalloc(sizeof(*zs));
robj *o;
zs->dict = dictCreate(&zsetDictType,NULL);
zs->zsl = zslCreate();
o = createObject(OBJ_ZSET,zs);
o->encoding = OBJ_ENCODING_SKIPLIST;
return o;
}
雜湊表和跳錶的資料必須保持一致。我們通過 zsetAdd 函式研究一下。
zsetAdd
啥都不說了,都在流程圖裡。
首先判斷編碼是 ziplist,還是 skiplist。
ziplist 編碼
裡面需要判斷是否要轉換編碼,如果轉換編碼,則需要呼叫 zsetConvert 轉換成 ziplist 編碼,這裡就不敘述了。
// ziplist 編碼時的處理邏輯
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr;
// zset 存在要插入的元素
if ((eptr = zzlFind(zobj->ptr, ele, &curscore)) != NULL) {
// 儲存要插入的元素時,在 not exist 時更新
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
……
if (newscore) *newscore = score;
// 原來的 score 和待插入 score 不同
if (score != curscore) {
// 先刪除原來的元素
zobj->ptr = zzlDelete(zobj->ptr, eptr);
// 插入新元素
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// zset 中不存在要插入的元素
else if (!xx) {
// 檢測 ele 是否過大 || 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 {
// 在 ziplist 中插入 (element,score) pair
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 編碼
// skiplist 編碼時的處理邏輯
if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
// 從雜湊表中查詢新增元素
de = dictFind(zs->dict, ele);
// 查詢到該元素
if (de != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
……
if (newscore) *newscore = score;
// 權重發生變化
if (score != curscore) {
// 更新跳錶節點
znode = zslUpdateScore(zs->zsl, curscore, ele, score);
// 讓雜湊表的元素的值指向跳錶節點的權重
dictGetVal(de) = &znode->score; /* Update score ptr. */
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// 如果新元素不存在
else if (!xx) {
ele = sdsdup(ele);
// 在跳錶中插入新元素
znode = zslInsert(zs->zsl, score, ele);
// 在雜湊表中插入新元素
serverAssert(dictAdd(zs->dict, ele, &znode->score) == DICT_OK);
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
}
zsetAdd 整體程式碼
int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
/* Turn options into simple to check vars. */
int incr = (in_flags & ZADD_IN_INCR) != 0;
int nx = (in_flags & ZADD_IN_NX) != 0;
int xx = (in_flags & ZADD_IN_XX) != 0;
int gt = (in_flags & ZADD_IN_GT) != 0;
int lt = (in_flags & ZADD_IN_LT) != 0;
*out_flags = 0; /* We'll return our response flags. */
double curscore;
/* NaN as input is an error regardless of all the other parameters. */
// 判斷 score 是否合法,不合法直接 return
if (isnan(score)) {
*out_flags = ZADD_OUT_NAN;
return 0;
}
/* Update the sorted set according to its encoding. */
// ziplist 編碼時的處理邏輯
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr;
// zset 存在要插入的元素
if ((eptr = zzlFind(zobj->ptr, ele, &curscore)) != NULL) {
// 儲存要插入的元素時,在 not exist 時更新
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
/* Prepare the score for the increment if needed. */
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
}
/* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
if (newscore) *newscore = score;
// 原來的 score 和待插入 score 不同
if (score != curscore) {
// 先刪除原來的元素
zobj->ptr = zzlDelete(zobj->ptr, eptr);
// 插入新元素
zobj->ptr = zzlInsert(zobj->ptr, ele, score);
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// zset 中不存在要插入的元素
else if (!xx) {
// 檢測 ele 是否過大 || 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 {
// 在 ziplist 中插入 (element,score) pair
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;
}
}
/* Note that the above block handling ziplist would have either returned or
* converted the key to skiplist. */
// skiplist 編碼時的處理邏輯
if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
// 從雜湊表中查詢新增元素
de = dictFind(zs->dict, ele);
// 查詢到該元素
if (de != NULL) {
/* NX? Return, same element already exists. */
if (nx) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
// 從雜湊表中查詢元素的權重
curscore = *(double *) dictGetVal(de);
// 如果要更新元素權重值
if (incr) {
score += curscore;
if (isnan(score)) {
*out_flags |= ZADD_OUT_NAN;
return 0;
}
}
/* GT/LT? Only update if score is greater/less than current. */
if ((lt && score >= curscore) || (gt && score <= curscore)) {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
if (newscore) *newscore = score;
// 權重發生變化
if (score != curscore) {
// 更新跳錶節點
znode = zslUpdateScore(zs->zsl, curscore, ele, score);
// 讓雜湊表的元素的值指向跳錶節點的權重
dictGetVal(de) = &znode->score; /* Update score ptr. */
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
}
// 如果新元素不存在
else if (!xx) {
ele = sdsdup(ele);
// 在跳錶中插入新元素
znode = zslInsert(zs->zsl, score, ele);
// 在雜湊表中插入新元素
serverAssert(dictAdd(zs->dict, ele, &znode->score) == DICT_OK);
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* Never reached. */
}
zsetDel
int zsetDel(robj *zobj, sds ele) {
// ziplist 編碼
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *eptr;
// 找到對應的節點
if ((eptr = zzlFind(zobj->ptr, ele, NULL)) != NULL) {
// 從 ziplist 中刪除
zobj->ptr = zzlDelete(zobj->ptr, eptr);
return 1;
}
}
// skiplist 編碼
else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
// 從 skiplist 中刪除
if (zsetRemoveFromSkiplist(zs, ele)) {
if (htNeedsResize(zs->dict)) dictResize(zs->dict);
return 1;
}
} else {
serverPanic("Unknown sorted set encoding");
}
return 0; /* No such element found. */
}
zsetRemoveFromSkiplist 函式如下:
static int zsetRemoveFromSkiplist(zset *zs, sds ele) {
dictEntry *de;
double score;
de = dictUnlink(zs->dict,ele);
if (de != NULL) {
score = *(double*)dictGetVal(de);
// 從雜湊表 unlink 該元素
dictFreeUnlinkedEntry(zs->dict,de);
// 從跳錶中刪除該元素,並釋放記憶體空間
int retval = zslDelete(zs->zsl,score,ele,NULL);
serverAssert(retval);
return 1;
}
return 0;
}
程式碼中的 zslDelete 函式在跳錶中分析過(文章中的跳錶章節)。
參考連結
Redis 原始碼簡潔剖析系列
Java 程式設計思想-最全思維導圖-GitHub 下載連結,需要的小夥伴可以自取~
原創不易,希望大家轉載時請先聯絡我,並標註原文連結。