2. 程式人生 > >Redis原始碼剖析--跳躍表zskiplist

Redis原始碼剖析--跳躍表zskiplist

阿新 • • 發佈：2019-01-30
跳躍表是一種有序的資料結構，它通過在每個節點中維持多個指向其他節點的指標，從而達到快速訪問的目的。跳躍表在插入、刪除和查詢操作上的平均複雜度為O（logN），最壞為O（N），可以和紅黑樹相媲美，但是在實現起來，比紅黑樹簡單很多。 > 說起跳躍表，在前段時間面試中可幫了我的大忙。騰訊一面的時候面試官要求設計一個數據結構，裡面的元素要求按一定順序存放，能以最低的複雜度獲取每個元素的名次，且增、刪等操作的複雜度儘可能低，博主最終就是用跳躍表來解決了這個問題，平均複雜度能達到O（logN）。 # 跳躍表資料結構 跳躍表的結構體定義在server.h檔案中。其中包括跳躍表節點zskiplistNode和跳躍表zskiplist兩個結構體。 
typedef struct zskiplistNode {
    robj *obj; // 成員物件
    double score;  // 分值
    struct zskiplistNode *backward; // 後向指標
    // 層
    struct zskiplistLevel {
        struct zskiplistNode *forward; // 前向指標
        unsigned int span; // 跨度
    } level[];
} zskiplistNode;

typedef struct zskiplist {
    // 跳躍表的表頭節點和表尾節點
 

    struct zskiplistNode *header, *tail;
    // 表中節點的數量
    unsigned long length;
    // 表中層數最大的節點層數
    int level;
} zskiplist;
對於跳躍表節點來說： + obj 存放著該節點對於的成員物件，一般指向一個sds結構 + score 表示該節點你的分值，跳躍表按照分值大小進行順序排列 + backward 指向跳躍表的前一個節點 + level[] 這個屬性至關重要，是跳躍表的核心所在，初始化一個跳躍表節點的時候會為其隨機生成一個層大小，每個節點的每一層以連結串列的形式連線起來。 看完上面的解釋之後，可能讀者對跳躍表還沒有一個清晰的認識，下面我畫了一張圖來形象的描述一下跳躍表結構。
# 跳躍表基本操作 Redis中關於跳躍表的相關操作函式定義在t_zset.c檔案中，下面分別介紹幾個基本操作函式的實現原始碼。 ## 建立跳躍表 Redis在建立一個跳躍表的時候完成以下操作： + 建立一個zskiplist結構 + 設定其level為1，長度length為0 + 初始化一個表頭結點，其層數為32層，每一層均指向NULL 
// 建立跳躍表
zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 申請記憶體
    zsl = zmalloc(sizeof(*zsl));
    // 初始化跳躍表屬性
 

    zsl->level = 1;
    zsl->length = 0;
    // 建立一個層數為32，分值為0，成員物件為NULL的表頭結點
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    // 設定每層的forward指標指向NULL
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    // 設定backward指向NULL
    zsl->header->backward = NULL;
    zsl->tail = NULL;
    return zsl;
}
// 建立一個跳躍表節點
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
    // 申請記憶體
    zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
    // 設定分值
    zn->score = score;
    // 設定成員物件
    zn->obj = obj;
    return zn;
}
## 插入節點 往跳躍表中插入一個節點，必然會改變跳錶的長度，可能會改變其長度。而且對於插入位置處的前後節點的backward和forward指標均要改變。 插入節點的關鍵在找到在何處插入該節點，跳躍表是按照score分值進行排序的，其查詢步驟大致是：從當前最高的level開始，向前查詢，如果當前節點的score小於插入節點的score，繼續向前；反之，則降低一層繼續查詢，直到第一層為止。此時，插入點就位於找到的節點之後。 
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    // updata[]陣列記錄每一層位於插入節點的前一個節點
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    // rank[]記錄每一層位於插入節點的前一個節點的排名
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;

    serverAssert(!isnan(score));
    x = zsl->header; // 表頭節點
    // 從最高層開始查詢
    for (i = zsl->level-1; i >= 0; i--) {
        // 儲存rank值是為了交叉快速地到達插入位置
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 前向指標不為空，前置指標的分值小於score或當前向指標的分值等// 於空但成員物件不等於o的情況下，繼續向前查詢
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                (x->level[i].forward->score == score &&
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            rank[i] += x->level[i].span;
            x = x->level[i].forward;
        }
        // 儲存當前層上位於插入節點的前一個節點
        update[i] = x;
    }
    // 此處假設插入節點的成員物件不存在於當前跳躍表內，即不存在重複的節點
    // 隨機生成一個level值
    level = zslRandomLevel();
    if (level > zsl->level) {
        // 如果level大於當前儲存的最大level值
        // 設定rank陣列中大於原level層以上的值為0
        // 同時設定update陣列大於原level層以上的資料
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        // 更新level值
        zsl->level = level;
    }
    // 建立插入節點
    x = zslCreateNode(level,score,obj);
    for (i = 0; i < level; i++) {
        // 針對跳躍表的每一層，改變其forward指標的指向
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x;

        // 更新插入節點的span值
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 更新插入點的前一個節點的span值
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }

    // 更新高層的span值
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 設定插入節點的backward指標
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    // 跳躍表長度+1
    zsl->length++;
    return x;
}
## 跳躍表刪除 Redis提供了三種跳躍表節點刪除操作。分別如下： + 根據給定分值和成員來刪除節點，由zslDelete函式實現 + 根據給定分值來刪除節點，由zslDeleteByScore函式實現 + 根據給定排名來刪除節點，由zslDeleteByRank函式實現 上述三種操作的刪除節點部分都由zslDeleteNode函式完成。zslDeleteNode函式用於刪除某個節點，需要給定當前節點和每一層下當前節點的前一個節點。 
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {
        if (update[i]->level[i].forward == x) {
            // 如果x存在於該層，則需要修改前一個節點的前向指標
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            // 反之，則只需要將span-1
            update[i]->level[i].span -= 1;
        }
    }
    // 修改backward指標，需要考慮x是否為尾節點
    if (x->level[0].forward) {
        x->level[0].forward->backward = x->backward;
    } else {
        zsl->tail = x->backward;
    }
    // 如果被刪除的節點為當前層數最多的節點，
    while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
        zsl->level--;
    zsl->length--;
}
以zslDelete為例，根據節點的分值和成員來刪除該節點，其他兩個操作無非是在查詢節點上有區別。 
int zslDelete(zskiplist *zsl, double score, robj *obj) {
    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 &&
                compareStringObjects(x->level[i].forward->obj,obj) < 0)))
            x = x->level[i].forward;
        update[i] = x;
    }
    // 跳躍表中可能存在分值相同的節點
    // 所以此處需要判斷成員是否相等
    x = x->level[0].forward;
    if (x && score == x->score && equalStringObjects(x->obj,obj)) {
        // 呼叫底層刪除節點函式
        zslDeleteNode(zsl, x, update);
        zslFreeNode(x);
        return 1;
    }
    // 沒有刪除成功
    return 0; 
}
## 獲取給定分值和成員的節點的排名 開篇提到博主在騰訊一面中被問的問題，需要獲取每個玩家的排名，跳躍表獲取排名的平均複雜度為O（logN），最壞為O（n）。其實現如下： 
unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
    zskiplistNode *x;
    unsigned long rank = 0;
    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 &&
                compareStringObjects(x->level[i].forward->obj,o) <= 0))) {
            // 前向指標不為空，前置指標的分值小於score或當前向指標的// 分值等於空但成員物件不等於o的情況下，繼續向前查詢
            rank += x->level[i].span;
            x = x->level[i].forward;
        }

        // 此時x可能是header，所以此處需要判斷一下
        if (x->obj && equalStringObjects(x->obj,o)) {
            return rank;
        }
    }
    return 0;
}
這裡粗略的畫了一張圖來說明查詢過程，紅線代表查詢的路線。

區間操作

Redis提供了一些區間操作，用於獲取某段區間上的節點或者刪除某段區間上的所有節點等操作，這些操作大大提高了Redis的易用性。

// 獲取某個區間上第一個符合範圍的節點。
zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;

    // 判斷給定的分值範圍是否在跳躍表的範圍內
    if (!zslIsInRange(zsl,range)) return NULL;

    x = zsl->header;

    for (i = zsl->level-1; i >= 0; i--) {
        // 如果當前節點的分值小於給定範圍的下限則一直向前查詢
        while (x->level[i].forward &&
            !zslValueGteMin(x->level[i].forward->score,range))
                x = x->level[i].forward;
    }

    // x的下一個節點才是我們要找的節點
    x = x->level[0].forward;
    serverAssert(x != NULL);

    // 檢查該節點不超過給定範圍範圍
    if (!zslValueLteMax(x->score,range)) return NULL;
    return x;
}
// 獲取某個區間上最後一個符合範圍的節點。
zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;

    // 判斷給定的分值範圍是否在跳躍表的範圍內
    if (!zslIsInRange(zsl,range)) return NULL;

    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        // 如果在給定範圍內則一直向前查詢
        while (x->level[i].forward &&
            zslValueLteMax(x->level[i].forward->score,range))
                x = x->level[i].forward;
    }

    // x即為要找的節點
    serverAssert(x != NULL);

    // 判斷該分值是否在給定範圍內
    if (!zslValueGteMin(x->score,range)) return NULL;
    return x;
}
// 刪除給定分值範圍內的所有元素
unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec *range, dict *dict) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long removed = 0;
    int i;

    x = zsl->header;
    // 找到小於或等於給定範圍最小分值的節點
    // 並將每層上的節點儲存到update陣列
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (range->minex ?
            x->level[i].forward->score <= range->min :
            x->level[i].forward->score < range->min))
                x = x->level[i].forward;
        update[i] = x;
    }

    // x的下一個節點則是給定區間內分值最小的節點
    x = x->level[0].forward;

    // 刪除該區間下的所有節點
    while (x &&
           (range->maxex ? x->score < range->max : x->score <= range->max))
    {
        // 儲存下一個節點
        zskiplistNode *next = x->level[0].forward;
        // 刪除該節點
        zslDeleteNode(zsl,x,update);
        // 刪除該節點的成員
        dictDelete(dict,x->obj);
        // 釋放該節點
        zslFreeNode(x);
        removed++;
        x = next;
    }
    // 返回刪除節點的個數
    return removed;
}
// 刪除給定排名區間內的所有節點
unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long traversed = 0, removed = 0;
    int i;

    x = zsl->header;
    // 找到給定排名區間內名次最小的節點
    // 並儲存每一層下該節點的前一個節點
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward && (traversed + x->level[i].span) < start) {
            traversed += x->level[i].span;
            x = x->level[i].forward;
        }
        update[i] = x;
    }
    // traversed儲存當前刪除節點的排名值
    traversed++;
    x = x->level[0].forward;
    while (x && traversed <= end) {
        // 記錄下一個節點
        zskiplistNode *next = x->level[0].forward;
        // 刪除該節點
        zslDeleteNode(zsl,x,update);
        // 刪除該節點的成員
        dictDelete(dict,x->obj);
        // 釋放該節點
        zslFreeNode(x);
        // 個數+1
        removed++;
        // 排名值加1
        traversed++;
        x = next;
    }
    // 返回刪除的節點個數
    return removed;
}

跳躍表小結

跳躍表是有序集合的底層實現之一。在同一個跳躍表中，多個節點可以包含相同的分值，但每個節點的成員物件必須是唯一的。跳躍表的節點是按照分值進行排序的，當分值相同時，節點按照成員物件的大小進行排序。

相關推薦

Redis原始碼剖析--跳躍表zskiplist

跳躍表是一種有序的資料結構，它通過在每個節點中維持多個指向其他節點的指標，從而達到快速訪問的目的。跳躍表在插入、刪除和查詢操作上的平均複雜度為O（logN），最壞為O（N），可以和紅黑樹相媲美，但是在實現起來，比紅黑樹簡單很多。 > 說起跳躍表，在前

redis原始碼之跳躍表skiplist

未完待續… 跳躍表skiplist 1.簡介： 離散，有序的資料結構。和連結串列、字典等資料結構被廣泛地應用在Redis 內部不同， Redis 只在兩個地方用到了跳躍表，一個是實現有序集

唯快不破：redis原始碼剖析03-dict雜湊表結構

精讀了一遍dict整個實現的原始碼，發現dict的實現程式碼還是比較優秀的，程式碼邏輯非常嚴謹。尤其是dict的擴充實現感覺比較好，dict每次擴充都是按照2的指數被擴充，在擴充的過程是循序漸進的，不是一下全部由老表遷移到新表，而是有個再hash的過程，會將以前的元素重新再

Redis原始碼剖析--記憶體管理zmalloc

功能函式總覽 在zmalloc.h中，定義了Redis記憶體分配的主要功能函式，這些函式基本上實現了Redis記憶體申請，釋放和統計等功能，其函式宣告如下： void *zmalloc(size_t size); /

Redis原始碼剖析（二）--簡單動態字串

 Redis沒有使用C語言的字串結構，而是自己設計了一個簡單的動態字串結構sds。它的特點是：可動態擴充套件記憶體、二進位制安全和與傳統的C語言字串型別相容。下面就從原始碼的角度來分析一下Redis中sds的實現。   1 SDS的定義 // sds相容傳統C風格字串，所以起了

Redis原始碼剖析（七）--壓縮列表

壓縮列表的構成 壓縮列表是為了節省記憶體而開發的，可以包含任意多個節點(entry)，每一個節點可以儲存一個位元組陣列或者一個整數值。 下圖展示了壓縮列表的各個組成部分， zlbytes：4位元組，壓縮列表所用的位元組數 zltail：4位元組，記錄壓縮列表尾節點entryN距離壓縮列表的

Redis原始碼剖析和註釋（二十五）--- Redis Cluster 的通訊流程深入剖析（載入配置檔案、節點握手、分配槽）

Redis Cluster 通訊流程深入剖析 1. Redis Cluster 介紹和搭建 這篇部落格會介紹Redis Cluster的資料分割槽理論和一個三主三從叢集的搭建。 2. Redis Cluster 和 Redis Sentin

Redis原始碼剖析和註釋（二十七）--- Redis 故障轉移流程和原理剖析

Redis 故障轉移流程和原理 1. 故障轉移介紹 Redis叢集自身實現了高可用。高可用首先要解決叢集部分失敗的場景：當叢集內少量節點出現故障時通過自動故障轉移保證叢集可以正常對外提供服務。接下來就介紹故障轉移的細節，分析故障檢測和故障轉移。 故障檢測

Redis原始碼剖析（九）--資料庫

伺服器中的資料庫 資料庫的實現 Redis的所有資料庫儲存在redisServer結構的db陣列中，每個redisDb結構代表一個數據庫，redisServer結構中的dbnum變數決定了有多少個數據庫（預設為16個數據庫） struct redisServer { // 資料庫

Redis原始碼剖析（十）--RDB持久化

RDB觸發機制 命令觸發 SAVE：SAVE命令會阻塞Redis服務程序，知道RDB檔案建立完畢為止。 BGSAVE：BGSAVE會建立子程序，子程序負責建立RDB檔案，父程序繼續處理命令請求 自動間隔性儲存 當配置檔案中save選項的條件滿足時，伺服器自動執行BGSAVE命令。

Redis原始碼剖析和註釋（十四）---- Redis 資料庫及相關命令實現(db)

Redis 資料庫及相關命令實現 1. 資料庫管理命令 命令 描述 FLUSHDB 清空當前資料庫的所有key FLUSHALL 清空整個Redis伺服器的所有key DBSIZE 返回當前資料庫的

Redis原始碼剖析--整數集合Intset

本系列部落格文章已經分析了Redis的大部分資料結構，包括動態字串，雙端連結串列，字典，跳躍表等，這些資料結構都非常強大實用，但是在記憶體消耗方面也非常“巨大”。Redis的資料都是存放在記憶體上面的，所以對記憶體的使用要求及其苛刻，Redis會想方設法的來節省

Redis中的跳躍表

跳躍表   跳躍表(skiplist)是一種有序資料結構，它通過在每個節點中維持多個指向其他節點的指標，從而達到快速訪問節點的目的。   跳躍表支援平均O(logN)、最壞O(N)複雜度的節點查詢，還可以通過順序性操作來批量處理節點。   在大部分情況

Redis原始碼剖析——adlist的實現

adlist adlist為Redis基本資料結構之一，為雙向連結串列，記錄了連結串列長度，adlist的迭代器記錄了迭代節點和方向，個人覺得實現優於STL的list 幾個重要結構 adlist實

Redis原始碼剖析和註釋（二十）--- 網路連線庫剖析(client的建立/釋放、命令接收/回覆、Redis通訊協議分析等)

Redis 網路連線庫剖析 1. Redis網路連線庫介紹 Redis網路連線庫對應的檔案是networking.c。這個檔案主要負責 客戶端的建立與釋放 命令接收與命令回覆 Redis通訊協議分析 CLIENT 命令的實現 我們接下來就這幾塊內

Redis原始碼剖析和註釋（十九）--- Redis 事件處理實現

Redis 事件處理實現 1. Redis事件介紹 Redis伺服器是一個事件驅動程式。下面先來簡單介紹什麼是事件驅動。 所謂事件驅動，就是當你輸入一條命令並且按下回車，然後訊息被組裝成Redis協議的格式傳送給Redis伺服器，這就會產生一個事件，Red

Redis原始碼剖析--基數統計hyperloglog

Redis中hyperloglog是用來做基數統計的，其優點是：在輸入元素的數量或者體積非常非常大的時候，計算基數所需的空間總是固定的，並且是很小的。在Redis裡面，每個Hyperloglog鍵只需要12Kb的大小就能計算接近2^64個不同元素的基數，但是hy

Redis原始碼剖析和註釋（十一）--- 雜湊鍵命令的實現(t_hash)

Redis 雜湊鍵命令實現(t_hash) 1. 雜湊命令介紹 Redis 所有雜湊命令如下表所示：Redis 雜湊命令詳解 序號 命令及描述 1 HDEL key field2 [field2]：刪除一個或多個雜湊表字段

Redis原始碼剖析（五）訂閱與釋出

Redis提供了訂閱和釋出的功能，允許客戶端訂閱一個或多個頻道，當其他客戶端向某個頻道傳送訊息時，伺服器會將訊息轉發給所有訂閱該頻道的客戶端 這一點有點像群聊的功能，一個客戶端將訊息發往群中(向某個頻道傳送訊息)，所有在群中的客戶端(訂閱該頻道的客戶端)都會收

Redis原始碼剖析--雙端連結串列Sdlist

> 請持續關注我的個人部落格：https://zcheng.ren 今天來分析Redis的一個基本資料結構–雙端連結串列，其定義和實現主要在sdlist.h和sdlist.c檔案中。其主要用在實現列表鍵、事務模組儲存輸入命令和伺服器模組，訂閱模組儲存多個