Redis 跳躍表(skiplist)

1. 跳躍表(skiplist)介紹

定義：跳躍表是一個(gè)有序鏈表，其中每個(gè)節(jié)點(diǎn)包含不定數(shù)量的鏈接，節(jié)點(diǎn)中的第i個(gè)鏈接構(gòu)成的單向鏈表跳過含有少于i個(gè)鏈接的節(jié)點(diǎn)。

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

跳躍表在redis中當(dāng)數(shù)據(jù)較多時(shí)作為有序集合鍵的實(shí)現(xiàn)方式之一。

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

127.0.0.1:6379> ZADD score 95.5 Mike 98 Li 96 Wang //socre是一個(gè)有序集合鍵(integer) 3127.0.0.1:6379> ZRANGE score 0 -1 WITHSCORES//所有分?jǐn)?shù)按從小到大排列，每一個(gè)成員都保存了一個(gè)分?jǐn)?shù)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. 跳躍表的實(shí)現(xiàn)

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

• 跳躍表節(jié)點(diǎn) zskiplistNode

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

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

typedef struct zskiplist { struct zskiplistNode *header, *tail;//header指向跳躍表的表頭節(jié)點(diǎn)，tail指向跳躍表的表尾節(jié)點(diǎn) unsigned long length; //跳躍表的長度或跳躍表節(jié)點(diǎn)數(shù)量計(jì)數(shù)器，除去第一個(gè)節(jié)點(diǎn) int level; //跳躍表中節(jié)點(diǎn)的最大層數(shù)，除了第一個(gè)節(jié)點(diǎn)} zskiplist;

剛才score鍵的所包含的成員空間結(jié)構(gòu)可能如下：

3. 冪次定律

在redis中，返回一個(gè)隨機(jī)層數(shù)值，隨機(jī)算法所使用的冪次定律。

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

表現(xiàn)是：少數(shù)幾個(gè)事件的發(fā)生頻率占了整個(gè)發(fā)生頻率的大部分， 而其余的大多數(shù)事件只占整個(gè)發(fā)生頻率的一個(gè)小部分。

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

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

算法性能分析：

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

層數(shù)恰好等于2的概率為 p(1?p)p(1?p)(執(zhí)行1次while循環(huán)體)。

層數(shù)恰好等于3的概率為 p2(1?p)p2(1?p)(執(zhí)行2次while循環(huán)體)。

層數(shù)恰好等于4的概率為 p3(1?p)p3(1?p)(執(zhí)行3次while循環(huán)體)。

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

因此，一個(gè)節(jié)點(diǎn)的平均層數(shù)，或平均指針數(shù)為：

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)

因此，

當(dāng) p = 1/2 時(shí)，每個(gè)節(jié)點(diǎn)的平均指針為2；

當(dāng) p = 1/4 時(shí)，每個(gè)節(jié)點(diǎn)的平均指針為1.33；

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

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

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

跳躍表和平衡樹做范圍查找時(shí)，跳躍表算法簡單，實(shí)現(xiàn)方便，而平衡樹邏輯復(fù)雜。

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

跳躍表平均每個(gè)節(jié)點(diǎn)包含1.33個(gè)指針，而平衡樹每個(gè)節(jié)點(diǎn)包含2個(gè)指針，更加節(jié)約內(nèi)存。

因此，在redis中實(shí)現(xiàn)有序集合的辦法是：跳躍表+哈希表

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

哈希表查找單個(gè)key時(shí)間復(fù)雜度性能高。

4. 跳躍表基本操作

redis關(guān)于跳躍表的API都定義在t_zset.c文件中。

4.1 創(chuàng)建跳躍表 zslCreate()

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

4.1 插入節(jié)點(diǎn) zslInsert()

//創(chuàng)建一個(gè)節(jié)點(diǎn)，分?jǐn)?shù)為score，對(duì)象為obj，插入到zsl表頭管理的跳躍表中，并返回新節(jié)點(diǎn)的地址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; //獲取跳躍表頭結(jié)點(diǎn)地址，從頭節(jié)點(diǎn)開始一層一層遍歷 for (i = zsl->level-1; i >= 0; i--) { //遍歷頭節(jié)點(diǎn)的每個(gè)level，從下標(biāo)最大層減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所跨越的節(jié)點(diǎn)數(shù)，但是最外一層為0 //這個(gè)while循環(huán)是查找的過程，沿著x指針遍歷跳躍表，滿足以下條件則要繼續(xù)在當(dāng)層往前走 while (x->level[i].forward && //當(dāng)前層的前進(jìn)指針不為空且 (x->level[i].forward->score < score || //當(dāng)前的要插入的score大于當(dāng)前層的score或 (x->level[i].forward->score == score && //當(dāng)前score等于要插入的score且 compareStringObjects(x->level[i].forward->obj,obj) < 0))) {//當(dāng)前層的對(duì)象與要插入的obj不等 rank[i] += x->level[i].span; //記錄該層一共跨越了多少節(jié)點(diǎn) 加上 上一層遍歷所跨越的節(jié)點(diǎn)數(shù) x = x->level[i].forward; //指向下一個(gè)節(jié)點(diǎn) } //while循環(huán)跳出時(shí)，用update[i]記錄第i層所遍歷到的最后一個(gè)節(jié)點(diǎn)，遍歷到i=0時(shí)，就要在該節(jié)點(diǎn)后要插入節(jié)點(diǎn) 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() 的調(diào)用者會(huì)確保同分值且同成員的元素不會(huì)出現(xiàn)， * 所以這里不需要進(jìn)一步進(jìn)行檢查，可以直接創(chuàng)建新元素。 */ level = zslRandomLevel(); //獲得一個(gè)隨機(jī)的層數(shù) if (level > zsl->level) { //如果大于當(dāng)前所有節(jié)點(diǎn)最大的層數(shù)時(shí) for (i = zsl->level; i < level; i++) { rank[i] = 0; //將大于等于原來zsl->level層以上的rank[]設(shè)置為0 update[i] = zsl->header; //將大于等于原來zsl->level層以上update[i]指向頭結(jié)點(diǎn) update[i]->level[i].span = zsl->length; //update[i]已經(jīng)指向頭結(jié)點(diǎn)，將第i層的跨度設(shè)置為length //length代表跳躍表的節(jié)點(diǎn)數(shù)量 } zsl->level = level; //更新表中的最大成數(shù)值 } x = zslCreateNode(level,score,obj); //創(chuàng)建一個(gè)節(jié)點(diǎn) for (i = 0; i < level; i++) { //遍歷每一層 x->level[i].forward = update[i]->level[i].forward; //設(shè)置新節(jié)點(diǎn)的前進(jìn)指針為查找時(shí)(while循環(huán))每一層最后一個(gè)節(jié)點(diǎn)的的前進(jìn)指針 update[i]->level[i].forward = x;//再把查找時(shí)每層的最后一個(gè)節(jié)點(diǎn)的前進(jìn)指針設(shè)置為新創(chuàng)建的節(jié)點(diǎn)地址 /* update span covered by update[i] as x is inserted here */ x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]); //更新插入節(jié)點(diǎn)的跨度值 update[i]->level[i].span = (rank[0] - rank[i]) + 1; //更新插入節(jié)點(diǎn)前一個(gè)節(jié)點(diǎn)的跨度值 } /* increment span for untouched levels */ for (i = level; i < zsl->level; i++) { //如果插入節(jié)點(diǎn)的level小于原來的zsl->level才會(huì)執(zhí)行 update[i]->level[i].span++; //因?yàn)楦叨葲]有達(dá)到這些層，所以只需將查找時(shí)每層最后一個(gè)節(jié)點(diǎn)的值的跨度加1 } //設(shè)置插入節(jié)點(diǎn)的后退指針，就是查找時(shí)最下層的最后一個(gè)節(jié)點(diǎn)，該節(jié)點(diǎn)的地址記錄在update[0]中 //如果插入在第二個(gè)節(jié)點(diǎn)，也就是頭結(jié)點(diǎn)后的位置就將后退指針設(shè)置為NULL x->backward = (update[0] == zsl->header) ? NULL : update[0]; if (x->level[0].forward) //如果x節(jié)點(diǎn)不是最尾部的節(jié)點(diǎn) x->level[0].forward->backward = x; //就將x節(jié)點(diǎn)后面的節(jié)點(diǎn)的后退節(jié)點(diǎn)設(shè)置成為x地址 else zsl->tail = x; //否則更新表頭的tail指針，指向最尾部的節(jié)點(diǎn)x zsl->length++; //跳躍表節(jié)點(diǎn)計(jì)數(shù)器加1 return x; //返回x地址}

4.3 刪除節(jié)點(diǎn)

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

4.4 獲取節(jié)點(diǎn)排名

unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) { //查找score和o對(duì)象在跳躍表中的排位 zskiplistNode *x; unsigned long rank = 0; int i; x = zsl->header; //遍歷頭結(jié)點(diǎn)的每一層 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 && //分值相等但是對(duì)象小于給定對(duì)象o compareStringObjects(x->level[i].forward->obj,o) <= 0))) { rank += x->level[i].span; //更新排位值 x = x->level[i].forward; //指向下一個(gè)節(jié)點(diǎn) } /* x might be equal to zsl->header, so test if obj is non-NULL */ //確保在第i層找到分值相同，且對(duì)象相同時(shí)才會(huì)返回排位值 if (x->obj && equalStringObjects(x->obj,o)) { return rank; } } return 0; //沒找到}

4.5 區(qū)間操作

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。