2. 程式人生 > 其它 >redis原始碼閱讀-資料結構篇-跳錶

redis原始碼閱讀-資料結構篇-跳錶

阿新 • • 發佈：2021-06-10

@

目錄

4. 跳錶實現 redis.h 和 t_zset.c

資料結構定義

  • object

    // 包含 meta 和 資料 指標
#define REDIS_LRU_BITS 24
typedef struct redisObject {
    unsigned type:4;				// 型別
    unsigned encoding:4;  			// 編碼
    unsigned lru:REDIS_LRU_BITS; 	// 物件最後一次被訪問的時間
    int refcount;					// 引用計數
    void *ptr;						// 指向實際值的指標
} robj;

  • 跳錶節點 value 為 object*

    typedef struct zskiplistNode {
    robj *obj; 							// 成員物件
    double score; 						// 分值
    struct zskiplistNode *backward;		// 後退指標
    struct zskiplistLevel {
        struct zskiplistNode *forward; 	// 前進指標
        unsigned int span;				// 跨度
    } level[]; 							// 層
} zskiplistNode;

  • 跳錶

    // meta 和 首尾指標
typedef struct zskiplist {
    struct zskiplistNode *header, *tail;	// 表頭節點和表尾節點
    unsigned long length;					// 表中節點的數量
    int level;								// 表中層數最大的節點的層數
} zskiplist;

Helper函式（可跳過，需要時閱讀）

  • 生成隨機層數 O(1)

    // 介於 1 - ZSKIPLIST_MAXLEVEL
// 層數越大，生成概率越小
int zslRandomLevel(void) {
    int level = 1;
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

  • 範圍判斷O(1)

    // 資料結構
// 開區間/閉區間範圍
typedef struct {
    // 最小值和最大值
    double min, max;
    // 指示最小值和最大值是否*不*包含在範圍之內
    // 值為 1 表示不包含，值為 0 表示包含
    int minex, maxex; 
} zrangespec;

// 大於等於最小值
static int zslValueGteMin(double value, zrangespec *spec) {
    return spec->minex ? (value > spec->min) : (value >= spec->min);
}
// 小於等於最大值
static int zslValueLteMax(double value, zrangespec *spec) {
    return spec->maxex ? (value < spec->max) : (value <= spec->max);
}

// 此處往下的範圍函式，只展示其一
typedef struct {
    robj *min, *max;  
    int minex, maxex; 
} zlexrangespec;

static int zslLexValueGteMin(robj *value, zlexrangespec *spec) {
    return spec->minex ?
        (compareStringObjectsForLexRange(value,spec->min) > 0) :
        (compareStringObjectsForLexRange(value,spec->min) >= 0);
}

static int zslLexValueLteMax(robj *value, zlexrangespec *spec) {
    return spec->maxex ?
        (compareStringObjectsForLexRange(value,spec->max) < 0) :
        (compareStringObjectsForLexRange(value,spec->max) <= 0);
}

  • list 和 範圍是否有交集 O(1)

    // 全list score 範圍判斷，是否有交集
int zslIsInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    // 先排除總為空的範圍值
    if (range->min > range->max ||
            (range->min == range->max && (range->minex || range->maxex)))
        return 0;
    // list 按照 score 升序排序
    // 檢查最大分值
    x = zsl->tail;
    if (x == NULL || !zslValueGteMin(x->score,range))
        return 0;
    // 檢查最小分值
    x = zsl->header->level[0].forward;
    if (x == NULL || !zslValueLteMax(x->score,range))
        return 0;
    return 1;
}

  • 第一個範圍內的 node O(logn)

    zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;
    if (!zslIsInRange(zsl,range)) return NULL;
    // 遍歷跳躍表，查詢符合範圍 min 項的節點 T_wrost = O(N), T_avg = O(log N)
    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->level[0].forward;
    redisAssert(x != NULL);
    // 檢查節點是否符合範圍的 max 項
    if (!zslValueLteMax(x->score,range)) return NULL;
    return x;
}

  • 最後一個範圍內的 node O(logn)

    zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec *range) {
    zskiplistNode *x;
    int i;
    // 確保跳躍表中至少有一個節點符合 range 指定的範圍，
    if (!zslIsInRange(zsl,range)) return NULL;

    // 遍歷跳躍表，查詢符合範圍 max 項的節點 T_wrost = O(N), T_avg = O(log N)
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        /* Go forward while *IN* range. */
        while (x->level[i].forward &&
            zslValueLteMax(x->level[i].forward->score,range))
                x = x->level[i].forward;
    }
    redisAssert(x != NULL);
    // 檢查節點是否符合範圍的 min 項
    if (!zslValueGteMin(x->score,range)) return NULL;
    return x;
}

node 建構函式 O(1)

// 分配 skiplist node的記憶體，設定 meta
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;
}

建構函式 O(1)

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    zsl = zmalloc(sizeof(*zsl));	// 分配空間
    zsl->level = 1; 				// 設定高度和起始層數
    zsl->length = 0;
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL); // 初始化表頭節點，具有最大level，輔助節點不設值
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;	// 初始無值
    zsl->tail = NULL;				// 表尾初始為 NULL
    return zsl;
}

node 解構函式 O(1)

不負責 object 的析構，只減少引用計數

void zslFreeNode(zskiplistNode *node) {
    decrRefCount(node->obj);
    zfree(node);
}

解構函式 O(n)

只負責本身 和 所有 node 的析構

不負責 object 的析構，只減少引用計數

void zslFree(zskiplist *zsl) {
    // level 0 可遍歷所有節點
    zskiplistNode *node = zsl->header->level[0].forward, *next;
    zfree(zsl->header);		// 釋放表頭
    while(node) { 			// 釋放表中所有節點
        next = node->level[0].forward;
        zslFreeNode(node);
        node = next;
    }
    zfree(zsl);				// 釋放跳躍表結構
}

insert object O(logn)

zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
   	zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    x = zsl->header;	// 在各個層查詢節點的插入位置 T_wrost = O(N), T_avg = O(log N)
    for (i = zsl->level-1; i >= 0; i--) {
        // 如果 i 不是 zsl->level-1 層
        // 那麼 i 層的起始 rank 值為 i+1 層的 rank 值
        // 各個層的 rank 值一層層累積
        // 最終 rank[0] 的值加一就是新節點的前置節點的排位
        // rank[0] 會在後面成為計算 span 值和 rank 值的基礎
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 沿著前進指標遍歷跳躍表, T_wrost = O(N^2), T_avg = O(N log N)
        while (x->level[i].forward && 					// 非空
            (x->level[i].forward->score < score ||		// score 大 或者
                (x->level[i].forward->score == score && // score 相等 並且 obj 大
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            rank[i] += x->level[i].span; 				// 記錄沿途跨越了多少個節點
            x = x->level[i].forward;  					// 移動至下一指標
        }												// 直到 x 的 score 大 或者 obj 大
        update[i] = x; 	// 記錄將要和新節點相連線的節點
    }
    level = zslRandomLevel();   // 獲取一個隨機值作為新節點的層數
    if (level > zsl->level) {	// 出現新level，注意初始level為1
        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,obj); // 建立新節點
    for (i = 0; i < level; i++) {		// 將前面記錄的指標指向新節點，並做相應的設定
        x->level[i].forward = update[i]->level[i].forward;// 設定新節點的 forward 指標
        update[i]->level[i].forward = x;// 將沿途記錄的各個節點的 forward 指標指向新節點
        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++) {// 未接觸的節點的 span 值也需要增一，這些節點直接從表頭指向新節點
        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;
}

delete object O(logn)

// O（1）
// 確定的node，帶有 insert 時相同的 update陣列，僅刪除連結串列結構，不涉及記憶體
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
    int i;
    for (i = 0; i < zsl->level; i++) {  // 更新所有和被刪除節點 x 有關的節點的指標，解除它們之間的關係
        if (update[i]->level[i].forward == x) {
            update[i]->level[i].span += x->level[i].span - 1;
            update[i]->level[i].forward = x->level[i].forward;
        } else {
            update[i]->level[i].span -= 1;
        }
    }
    if (x->level[0].forward) {// 更新被刪除節點 x 的前進和後退指標
        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--;
}

// 尋找並刪除帶有 score 的 obj，釋放 node 記憶體
int zslDelete(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    int i;
    // 遍歷跳躍表，查詢目標節點，並記錄所有沿途節點， T_wrost = O(N), T_avg = O(log N)
    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 = x->level[0].forward;
    if (x && score == x->score && equalStringObjects(x->obj,obj)) {
        zslDeleteNode(zsl, x, update);
        zslFreeNode(x);
        return 1;
    } else {
        return 0; 
    }
    return 0; 
}

根據score，範圍刪除 O(logn)

unsigned long zslDeleteRangeByScore(zskiplist *zsl, zrangespec *range, dict *dict){
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned long removed = 0;
    int i;
    // 記錄所有和被刪除節點（們）有關的節點 T_wrost = O(N) , T_avg = O(log N)
    x = zsl->header;
    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->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;
}

根據rank，範圍刪除 O(logn)

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;
    // 找到start，並記錄所有沿途指標 T_wrost = O(N) , T_avg = O(log N)
    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++; 						// 移動到排位的起始的第一個節點
    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);					// 釋放節點結構
        removed++;						// 為刪除計數器增一
        traversed++;					// 為排位計數器增一
        x = next;						// 處理下個節點
    }
    return removed;						// 返回被刪除節點的數量
}

獲取 rank O(logn)

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))) {
            rank += x->level[i].span; 				// 累積跨越的節點數量
            x = x->level[i].forward;				// 沿著前進指標遍歷跳躍表
        }
        if (x->obj && equalStringObjects(x->obj,o)) {// 確認是否 equal，否則繼續
            return rank;
        }
    }
    return 0;
}

get by rank O(logn)

zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
    zskiplistNode *x;
    unsigned long traversed = 0;
    int i;
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        while (x->level[i].forward // 遍歷跳躍表並累積越過的節點數量
               && (traversed + x->level[i].span) <= rank){
            traversed += x->level[i].span;
            x = x->level[i].forward;
        }
        if (traversed == rank) {	// 找到節點
            return x;
        }
    }
    return NULL;				// 沒找到目標節點
}

相關推薦

redis原始碼閱讀-資料結構篇-跳錶

@目錄4. 跳錶實現 redis.h 和 t_zset.c資料結構定義Helper函式（可跳過，需要時閱讀）node 建構函式 O(1)建構函式 O(1)node 解構函式 O(1)解構函式 O(n)insert object O(logn)delete object O(logn)根據score，範圍刪

redis原始碼閱讀-資料結構篇-雙端列表adlist

目錄3. 雙端連結串列 adlist.h 和 adlist.c資料結構定義建構函式解構函式get iterdelete iterreset iter迭代add node to headinsert nodedelete nodeget node by keyget node by indexcopy listrotate

redis原始碼閱讀-資料結構篇-字典dict

@目錄4. 字典實現 dict.h 和 dict.c資料結構定義部分巨集定義雜湊函式（本部分可選跳過）Helper函式（可跳過，需要時閱讀）dict 建構函式 O(1)dict 解構函式 O(N)hashtable 建構函式 O(1)dict resize O(1)rehash O(m

redis原始碼閱讀-資料結構篇-hyperloglog

@目錄5. HyperLogLog 實現 hyperloglog.c資料結構定義Helper函式（可跳過，需要時閱讀）元素雜湊處理 O(1)新增基數 O(1)...

Redis的底層資料結構-跳錶

跳躍表（skiplist）是一種有序資料結構，它通過在每個節點中維持多個指向其它節點的指標，從而達到快速訪問節點的目的。具有如下性質：

redis原始碼閱讀—sds

技術標籤：redissds 文章目錄 sdshdr定義sds邏輯圖sds優勢更快的獲取字串長度杜絕緩衝區溢位減少修改字串時帶來的記憶體重分配次數二進位制安全

資料結構 - 跳錶

簡介 有序的陣列可以使用二分查詢的方法快速檢索一個數據，但是連結串列沒有辦法使用二分查詢。

Redis 6.0 原始碼閱讀筆記(8)- Stream 資料型別原始碼分析--TBD

文章目錄 1. 儲存結構2. 原始碼分析 1. 儲存結構 在 流物件 Stream 的介紹中已經提到 Stream 的底層儲存結構為字首壓縮樹，其結構示例如下：

Redis原始碼剖析之跳錶(skiplist)

最近要換新工作了，藉著新老工作交替的這段視窗放鬆了下，所以專欄拖更了，不過我心裡毫無愧疚，畢竟沒人催更。 不過話說回來天天追劇 刷綜藝的日子也很是枯燥，羨慕你們這些正常上班的人，每天都有正經工作內容，

資料結構與演算法_17 _ 跳錶：為什麼Redis一定要用跳錶來實現有序集合？

上兩節我們講了二分查詢演算法。當時我講到，因為二分查詢底層依賴的是陣列隨機訪問的特性，所以只能用陣列來實現。如果資料儲存在連結串列中，就真的沒法用二分查詢演算法了嗎？

Redis原始碼解析-基礎資料-ziplist(壓縮列表)

太長不看版 壓縮列表是一種為節約空間而實現的線性資料結構，本質上是位元組陣列。

Redis原始碼解析-基礎資料-skiplist(跳躍表)

太長不看版 跳躍表是有序集合zset的底層實現之一， 除此之外它在 Redis 中沒有其他應用。

簡單談談Mysql索引與redis跳錶

摘要 面試時，交流有關mysql索引問題時，發現有些人能夠濤濤不絕的說出B+樹和B樹，平衡二叉樹的區別，卻說不出B+樹和hash索引的區別。這種一看就知道是死記硬背，沒有理解索引的本質。本文旨在剖析這背後的原理，歡迎

連結串列加速器--淺談跳錶SkipList及在Redis中的應用

問題背景 眾所周知，對於一個有序集合中元素的查詢，如果集合是通過陣列儲存的，那麼利用陣列隨機訪問的高效性通過二分查詢演算法可以快速的查詢元素。那麼設想如果該集合是通過連結串列儲存，是否也能實現快速查

7.跳錶：各方面都非常優秀的動態資料結構

點選使用幕布網頁版檢視 跳錶基於單鏈表實現，連結串列查詢、插入、刪除絕大部分時間花在遍歷連結串列上，跳錶使用索引來優化連結串列遍歷的過程，使得跳錶具有非常優秀的查詢、插入、刪除效能，並且是天然的動態資

資料結構學習——skiplist跳錶

目錄 1.skiplist簡介 2.skiplist核心思想 3.skiplist原理 3.1 跳錶的查詢 3.2 跳錶的插入 3.3跳錶的刪除

Redis中的跳錶

date: 2020-10-15 14:58:00 updated: 2020-10-19 17:58:00 Redis中的跳錶 參考網址1 參考網址2 redis 資料型別 zset 實現有序集合，底層使用的資料結構是跳錶。

Redis6.0原始碼閱讀（一）redis-server啟動主流程

前言：redis進入到6.0之後開始支援多執行緒模式，並加入了許多新的功能，本系列是本人在學習過程中的一些記錄，希望能有更多的朋友能夠一起學習交流。

REDIS6.0原始碼閱讀（二）REDIS-SERVER啟動主流程之initServerConfig

在redis-server的初始化過程中，會從引數（redis.conf檔案中讀取使用者的配置資訊），還包含有一些諸如設定時區引數、設定記憶體溢位的回撥函式等，比較重要的是initServerConfig，initServerConfig主要包含有一些基

redis 5.0.7 原始碼閱讀——字典dict

redis中字典相關的檔案為：dict.h與dict.c 與其說是一個字典，道不如說是一個雜湊表。