Redis 源碼簡潔剖析 04 – Sorted Set 有序集合

2022 年 1 月 29 日
筆記
二進位文集

Sorted Set 是什麼
Sorted Set 命令及實現方法
Sorted Set 數據結構
跳錶（skiplist）
Sorted Set 基本操作
- zsetAdd
- zsetDel
參考鏈接
Redis 源碼簡潔剖析系列

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 源碼簡潔剖析系列

最簡潔的 Redis 源碼剖析系列文章

Java 編程思想-最全思維導圖-GitHub 下載鏈接，需要的小夥伴可以自取~

原創不易，希望大家轉載時請先聯繫我，並標註原文鏈接。

Tags: 二進位文集

Redis 源碼簡潔剖析 04 – Sorted Set 有序集合

Sorted Set 是什麼

Sorted Set 命令及實現方法

Sorted Set 數據結構

跳錶（skiplist）

跳錶節點的結構定義

跳錶的定義

跳錶節點查詢

層數設置

跳錶插入節點 zslInsert

跳錶刪除節點 zslDelete

Sorted Set 基本操作

zsetAdd

ziplist 編碼

skiplist 編碼

zsetAdd 整體程式碼

zsetDel

參考鏈接

Redis 源碼簡潔剖析系列

VirMach 便宜 VPS

QNews

Redis 源碼簡潔剖析 04 – Sorted Set 有序集合

Sorted Set 是什麼

Sorted Set 命令及實現方法

Sorted Set 數據結構

跳錶（skiplist）

跳錶節點的結構定義

跳錶的定義

跳錶節點查詢

層數設置

跳錶插入節點 zslInsert

跳錶刪除節點 zslDelete

Sorted Set 基本操作

zsetAdd

ziplist 編碼

skiplist 編碼

zsetAdd 整體程式碼

zsetDel

參考鏈接

Redis 源碼簡潔剖析系列

分享此文：

Related Posts

IOS小組件（7）：小組件點擊交互

演算法—BitMap

3DCenter：2022年顯示卡將會更便宜

港版華為nova 9、P50 Pro等手機將升級鴻蒙HarmonyOS 2

VirMach 便宜 VPS

QNews

熱門搜尋