Redis 跳躍表(skiplist)

1. 跳躍表(skiplist)介紹

定義：跳躍表是一個有序鏈表，其中每個節點包含不定數量的鏈接，節點中的第i個鏈接構成的單向鏈表跳過含有少于i個鏈接的節點。

跳躍表支持平均O(logN)，最壞O(N)復雜度的節點查找，大部分情況下，跳躍表的效率可以和平衡樹相媲美。

跳躍表在redis中當數據較多時作為有序集合鍵的實現方式之一。

接下來，還是舉個有序集合鍵的例子：

127.0.0.1:6379> ZADD score 95.5 Mike 98 Li 96 Wang //socre是一個有序集合鍵(integer) 3127.0.0.1:6379> ZRANGE score 0 -1 WITHSCORES//所有分數按從小到大排列，每一個成員都保存了一個分數1) "Mike"2) "95.5"3) "Wang"4) "96" 5) "Li"6) "98"127.0.0.1:6379> ZSCORE score Mike //查詢Mike的分值"95.5"

2. 跳躍表的實現

redis 3.0版本將跳躍表定義在redis.h文件中，而3.2版本定義在server.h文件中

• 跳躍表節點 zskiplistNode

typedef struct zskiplistNode { robj *obj; //保存成員對象的地址 double score; //分值 struct zskiplistNode *backward; //后退指針 struct zskiplistLevel { struct zskiplistNode *forward; //前進指針 unsigned int span; //跨度 } level[]; //層級，柔型數組} zskiplistNode;

• 跳躍表表頭 zskiplist(記錄跳躍表信息)

typedef struct zskiplist { struct zskiplistNode *header, *tail;//header指向跳躍表的表頭節點，tail指向跳躍表的表尾節點 unsigned long length; //跳躍表的長度或跳躍表節點數量計數器，除去第一個節點 int level; //跳躍表中節點的最大層數，除了第一個節點} zskiplist;

剛才score鍵的所包含的成員空間結構可能如下：

3. 冪次定律

在redis中，返回一個隨機層數值，隨機算法所使用的冪次定律。

含義是：如果某件事的發生頻率和它的某個屬性成冪關系，那么這個頻率就可以稱之為符合冪次定律。

表現是：少數幾個事件的發生頻率占了整個發生頻率的大部分， 而其余的大多數事件只占整個發生頻率的一個小部分。

在文件t_set.c中，zslRandomLevel函數的定義為：

int zslRandomLevel(void) { //返回一個隨機層數值 int level = 1; //(random()&0xFFFF)只保留低兩個字節的位值，其他高位全部清零，所以該值范圍為0到0xFFFF while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF)) //ZSKIPLIST_P(0.25)所以level+1的概率為0.25 level += 1; //返回一個1到ZSKIPLIST_MAXLEVEL(32)之間的值 return (level

算法性能分析：

層數至少為1，所以層數恰好等于1(不執行while循環體)的概率為 1?p1?p.

層數恰好等于2的概率為 p(1?p)p(1?p)(執行1次while循環體)。

層數恰好等于3的概率為 p2(1?p)p2(1?p)(執行2次while循環體)。

層數恰好等于4的概率為 p3(1?p)p3(1?p)(執行3次while循環體)。

層數恰好等于k的概率為 pk?1(1?p)pk?1(1?p)(執行k-1次while循環體)。(k <= ZSKIPLIST_MAXLEVEL)

因此，一個節點的平均層數，或平均指針數為：

1×(1?p)+2p(1?p)+3p2(1?p)+...+kpk?1(1?p)1×(1?p)+2p(1?p)+3p2(1?p)+...+kpk?1(1?p)

? = (1?p)∑+∞k=1kpk?1(1?p)∑k=1+∞kpk?1

? =(1?p)(1?p)1(1?p)21(1?p)2

? =1(1?p)1(1?p)

因此，

當 p = 1/2 時，每個節點的平均指針為2；

當 p = 1/4 時，每個節點的平均指針為1.33；

而redis的概率 ZSKIPLIST_P 取值就為0.25，所以跳躍表的指針開銷為1.33。

4. 跳躍表與哈希表和平衡樹的比較

跳躍表和平衡樹的元素都是有序排列，而哈希表不是有序的。因此在哈希表上的查找只能是單個key的查找，不適合做范圍查找。

跳躍表和平衡樹做范圍查找時，跳躍表算法簡單，實現方便，而平衡樹邏輯復雜。

查找單個key，跳躍表和平衡樹的平均時間復雜度都為O(logN)，而哈希表的時間復雜度為O(N)。

跳躍表平均每個節點包含1.33個指針，而平衡樹每個節點包含2個指針，更加節約內存。

因此，在redis中實現有序集合的辦法是：跳躍表+哈希表

跳躍表元素有序，而且可以范圍查找，且比平衡樹簡單。

哈希表查找單個key時間復雜度性能高。

4. 跳躍表基本操作

redis關于跳躍表的API都定義在t_zset.c文件中。

4.1 創建跳躍表 zslCreate()

zskiplist *zslCreate(void) { //創建返回一個跳躍表 表頭zskiplist int j; zskiplist *zsl; zsl = zmalloc(sizeof(*zsl)); //分配空間 zsl->level = 1; //設置默認層數 zsl->length = 0; //設置跳躍表長度 //創建一個層數為32，分數為0，沒有obj的跳躍表頭節點 zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL); //跳躍表頭節點初始化 for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) { zsl->header->level[j].forward = NULL; //將跳躍表頭節點的所有前進指針forward設置為NULL zsl->header->level[j].span = 0; //將跳躍表頭節點的所有跨度span設置為0 } zsl->header->backward = NULL; //跳躍表頭節點的后退指針backward置為NULL zsl->tail = NULL; //表頭指向跳躍表尾節點的指針置為NULL return zsl;}

4.1 插入節點 zslInsert()

//創建一個節點，分數為score，對象為obj，插入到zsl表頭管理的跳躍表中，并返回新節點的地址zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) { zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x; unsigned int rank[ZSKIPLIST_MAXLEVEL]; int i, level; redisAssert(!isnan(score)); x = zsl->header; //獲取跳躍表頭結點地址，從頭節點開始一層一層遍歷 for (i = zsl->level-1; i >= 0; i--) { //遍歷頭節點的每個level，從下標最大層減1到0 /* store rank that is crossed to reach the insert position */ rank[i] = i == (zsl->level-1) ? 0 : rank[i+1]; //更新rank[i]為i+1所跨越的節點數，但是最外一層為0 //這個while循環是查找的過程，沿著x指針遍歷跳躍表，滿足以下條件則要繼續在當層往前走 while (x->level[i].forward && //當前層的前進指針不為空且 (x->level[i].forward->score < score || //當前的要插入的score大于當前層的score或 (x->level[i].forward->score == score && //當前score等于要插入的score且 compareStringObjects(x->level[i].forward->obj,obj) < 0))) {//當前層的對象與要插入的obj不等 rank[i] += x->level[i].span; //記錄該層一共跨越了多少節點 加上 上一層遍歷所跨越的節點數 x = x->level[i].forward; //指向下一個節點 } //while循環跳出時，用update[i]記錄第i層所遍歷到的最后一個節點，遍歷到i=0時，就要在該節點后要插入節點 update[i] = x; } /* we assume the key is not already inside, since we allow duplicated * scores, and the re-insertion of score and redis object should never * happen since the caller of zslInsert() should test in the hash table * if the element is already inside or not. * zslInsert() 的調用者會確保同分值且同成員的元素不會出現， * 所以這里不需要進一步進行檢查，可以直接創建新元素。 */ level = zslRandomLevel(); //獲得一個隨機的層數 if (level > zsl->level) { //如果大于當前所有節點最大的層數時 for (i = zsl->level; i < level; i++) { rank[i] = 0; //將大于等于原來zsl->level層以上的rank[]設置為0 update[i] = zsl->header; //將大于等于原來zsl->level層以上update[i]指向頭結點 update[i]->level[i].span = zsl->length; //update[i]已經指向頭結點，將第i層的跨度設置為length //length代表跳躍表的節點數量 } zsl->level = level; //更新表中的最大成數值 } x = zslCreateNode(level,score,obj); //創建一個節點 for (i = 0; i < level; i++) { //遍歷每一層 x->level[i].forward = update[i]->level[i].forward; //設置新節點的前進指針為查找時(while循環)每一層最后一個節點的的前進指針 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; //更新插入節點前一個節點的跨度值 } /* increment span for untouched levels */ for (i = level; i < zsl->level; i++) { //如果插入節點的level小于原來的zsl->level才會執行 update[i]->level[i].span++; //因為高度沒有達到這些層，所以只需將查找時每層最后一個節點的值的跨度加1 } //設置插入節點的后退指針，就是查找時最下層的最后一個節點，該節點的地址記錄在update[0]中 //如果插入在第二個節點，也就是頭結點后的位置就將后退指針設置為NULL x->backward = (update[0] == zsl->header) ? NULL : update[0]; if (x->level[0].forward) //如果x節點不是最尾部的節點 x->level[0].forward->backward = x; //就將x節點后面的節點的后退節點設置成為x地址 else zsl->tail = x; //否則更新表頭的tail指針，指向最尾部的節點x zsl->length++; //跳躍表節點計數器加1 return x; //返回x地址}

4.3 刪除節點

//被zslDelete, zslDeleteByScore and zslDeleteByRank使用的內部函數void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) { //刪除節點 int i; //設置前進指針和跨度 for (i = 0; i < zsl->level; i++) { //遍歷下標為0到跳躍表最大層數-1的層 if (update[i]->level[i].forward == x) { //如果找到該節點 update[i]->level[i].span += x->level[i].span - 1; //將前一個節點的跨度減1 update[i]->level[i].forward = x->level[i].forward; //前一個節點的前進指針指向被刪除的節點的后一個節點，跳過該節點 } else { update[i]->level[i].span -= 1; //在第i層沒找到，只將該層的最后一個節點的跨度減1 } } //設置后退指針 if (x->level[0].forward) { //如果被刪除的前進節點不為空，后面還有節點 x->level[0].forward->backward = x->backward; //就將后面節點的后退指針指向被刪除節點x的回退指針 } else { zsl->tail = x->backward; //否則直接將被刪除的x節點的后退節點設置為表頭的tail指針 } //更新跳躍表最大層數 while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL) zsl->level--; zsl->length--; //節點計數器減1}

4.4 獲取節點排名

unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) { //查找score和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 || //只要分值還小于給定的score或者 (x->level[i].forward->score == score && //分值相等但是對象小于給定對象o compareStringObjects(x->level[i].forward->obj,o) <= 0))) { rank += x->level[i].span; //更新排位值 x = x->level[i].forward; //指向下一個節點 } /* x might be equal to zsl->header, so test if obj is non-NULL */ //確保在第i層找到分值相同，且對象相同時才會返回排位值 if (x->obj && equalStringObjects(x->obj,o)) { return rank; } } return 0; //沒找到}

4.5 區間操作

zskiplistNode *zslFirstInRange(zskiplist *zsl, zrangespec *range) { //返回第一個分數在range范圍內的節點 zskiplistNode *x; int i; /* If everything is out of range, return early. */ if (!zslIsInRange(zsl,range)) return NULL; //如果不在范圍內，則返回NULL，確保至少有一個節點符號range //判斷下限 x = zsl->header;//遍歷跳躍表 for (i = zsl->level-1; i >= 0; i--) {//遍歷每一層 /* Go forward while *OUT* of range. */ while (x->level[i].forward && //如果該層有下一個節點且 !zslValueGteMin(x->level[i].forward->score,range))//當前節點的score還小于(小于等于)range的min x = x->level[i].forward; //繼續指向下一個節點 } /* This is an inner range, so the next node cannot be NULL. */ x = x->level[0].forward; //找到目標節點 redisAssert(x != NULL); //保證能找到 /* Check if score <= max. */ //判斷上限 if (!zslValueLteMax(x->score,range)) return NULL; //該節點的分值如果比max還要大，就返回NULL return x;}zskiplistNode *zslLastInRange(zskiplist *zsl, zrangespec *range) {//返回最后一個分數在range范圍內的節點 zskiplistNode *x; int i; /* If everything is out of range, return early. */ if (!zslIsInRange(zsl,range)) return NULL; //如果不在范圍內，則返回NULL，確保至少有一個節點符號range //判斷上限 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))//當前節點的score小于(小于等于)max x = x->level[i].forward; //繼續指向下一個節點 } /* This is an inner range, so this node cannot be NULL. */ redisAssert(x != NULL);//保證能找到 /* Check if score >= min. */ //判斷下限 if (!zslValueGteMin(x->score,range)) return NULL; //如果找到的節點的分值比range的min還要小 return x;}

總結

以上是生活随笔為你收集整理的arcengine遍历属性表_Redis源码解析四--跳跃表的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。