Redis源码剖析(十一)跳表
在樹(shù)形結(jié)構(gòu)中,常見(jiàn)的平衡樹(shù)有AVL樹(shù)和紅黑樹(shù),但是由于AVL樹(shù)過(guò)于平衡,導(dǎo)致維護(hù)平衡所需的代價(jià)過(guò)大,使用的不多,不過(guò)其中幾種旋轉(zhuǎn)算法還是值得學(xué)習(xí)的。取而代之的是較為平衡的紅黑樹(shù),STL中的map和set都是采用紅黑樹(shù)實(shí)現(xiàn)的,插入和查找效率為O(logN)。
而跳表也是一種較為平衡的數(shù)據(jù)結(jié)構(gòu),與紅黑樹(shù)不同的是,它是鏈狀結(jié)構(gòu)而非樹(shù)形結(jié)構(gòu),不過(guò),跳表的插入查找效率也為O(logN),和紅黑樹(shù)有一拼,而且最重要的是,跳轉(zhuǎn)在實(shí)現(xiàn)上比紅黑樹(shù)簡(jiǎn)單的多
跳表結(jié)構(gòu)
同數(shù)組相比,鏈表的插入刪除效率是O(1),但是如果想要在鏈表中查找某個(gè)元素,就糟糕了,復(fù)雜度會(huì)是O(N),為了提高查找效率,就有了跳表的概念。所謂跳表,就是可以跳躍的鏈表,回想二分查找算法,每次的查找都是跳躍性的,這才使得二分法效率這么高,跳表的設(shè)計(jì)同樣也借鑒了二分法的策略,實(shí)現(xiàn)跳躍查找,當(dāng)然,需要跳表中的元素有序
普通的鏈表每個(gè)節(jié)點(diǎn)僅僅保存了指向下一個(gè)節(jié)點(diǎn)的指針,只能移動(dòng)到下一個(gè)相鄰節(jié)點(diǎn),也就是跳一步。而跳表為了可以一次跳很多步,保存了很多指針,指向該節(jié)點(diǎn)后面的不同節(jié)點(diǎn)
在server.h頭文件中,可以找到跳表節(jié)點(diǎn)的定義和跳表的定義
/* 跳表節(jié)點(diǎn) */ typedef struct zskiplistNode {robj *obj; /* 數(shù)據(jù) */double score; /* 分?jǐn)?shù) */struct zskiplistNode *backward; //前一個(gè)節(jié)點(diǎn)指針struct zskiplistLevel {struct zskiplistNode *forward; //后面某個(gè)節(jié)點(diǎn),也就是next指針unsigned int span; //跨度} level[]; /* 跳表中保存了多個(gè)指向下一個(gè)節(jié)點(diǎn)的指針 */ } zskiplistNode;跳表實(shí)際上也是保存鍵值對(duì)的結(jié)構(gòu),其中obj保存實(shí)際的數(shù)據(jù)而score用于排序使用,這保證了跳表內(nèi)部是有序的。此外,level數(shù)組記錄了多個(gè)指向后面節(jié)點(diǎn)的指針,同時(shí)也記錄了兩個(gè)節(jié)點(diǎn)之間的跨度
為了方便,接下來(lái)將跳表節(jié)點(diǎn)中指向下一個(gè)節(jié)點(diǎn)的指針?lè)Q為next指針,而level數(shù)組稱為next數(shù)組
/* 跳表 */ typedef struct zskiplist {struct zskiplistNode *header, *tail; //表頭表尾unsigned long length; /* 跳表中節(jié)點(diǎn)個(gè)數(shù) */int level; //跳表總層數(shù) } zskiplist;跳表定義中記錄了表頭表尾,level記錄了當(dāng)前跳表的總層數(shù)
下面是跳表的一個(gè)例子,可以看到,每個(gè)節(jié)點(diǎn)都有若干個(gè)next指針,通過(guò)這些指針,可以直線跳躍移動(dòng),而不再是只能移動(dòng)到相鄰節(jié)點(diǎn)
圖片轉(zhuǎn)自https://zcheng.ren/2016/12/06/TheAnnotatedRedisSourceZskiplist/
跳表操作
創(chuàng)建跳表
創(chuàng)建跳表由zslCreate函數(shù)實(shí)現(xiàn),函數(shù)中需要調(diào)用zslCreateNode創(chuàng)建跳表節(jié)點(diǎn),主要就是申請(qǐng)內(nèi)存,設(shè)置初值
//t_zset.c /*** 創(chuàng)建一個(gè)跳表節(jié)點(diǎn)* level : 節(jié)點(diǎn)包含的層數(shù),即節(jié)點(diǎn)next數(shù)組的大小,每一層有一個(gè)next指針 * score : 該節(jié)點(diǎn)的分值,用于使跳表數(shù)據(jù)有序* obj : 跳表保存的數(shù)據(jù)**/ zskiplistNode *zslCreateNode(int level, double score, robj *obj) {/* 申請(qǐng)?zhí)砉?jié)點(diǎn)內(nèi)存 */zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));/* 設(shè)置初值 */zn->score = score;zn->obj = obj;return zn; }/* 創(chuàng)建一個(gè)跳表 */ zskiplist *zslCreate(void) {int j;zskiplist *zsl;/* 申請(qǐng)?zhí)韮?nèi)存, 初始總層數(shù)為1 */zsl = zmalloc(sizeof(*zsl));zsl->level = 1;zsl->length = 0;/* 申請(qǐng)表頭節(jié)點(diǎn),默認(rèn)有32層 */zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);/* 對(duì)每一層進(jìn)行初始化 */for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {zsl->header->level[j].forward = NULL;zsl->header->level[j].span = 0;}zsl->header->backward = NULL;/* 設(shè)置表尾節(jié)點(diǎn) */zsl->tail = NULL;return zsl; }值得注意的是,表頭節(jié)點(diǎn)是在創(chuàng)建跳表就申請(qǐng)好的,不屬于跳表數(shù)據(jù)的一部分。
搜索操作
跳表的多數(shù)操作都是基于搜索的,考慮這樣一個(gè)需求,就是在跳表中查找某個(gè)節(jié)點(diǎn),如果該節(jié)點(diǎn)不存在,找到它應(yīng)該插入的位置。雖然Redis中沒(méi)有實(shí)現(xiàn)搜索功能,但是后面會(huì)看到,插入刪除等函數(shù)都是基于搜索的
假設(shè)此時(shí)跳表結(jié)構(gòu)如下,需要從跳表中查找分值為23的節(jié)點(diǎn)
查找操作的步驟如下
- 從最高層開(kāi)始嘗試移動(dòng),比較next節(jié)點(diǎn)的分值和待查找分值大小
- 如果next節(jié)點(diǎn)的分值小于待查找分值,則移動(dòng)到next節(jié)點(diǎn)
- 如果next節(jié)點(diǎn)的分值大于等于待查找分值,則降層,如果此時(shí)為第0層不能繼續(xù)降層,當(dāng)前節(jié)點(diǎn)位置就是待查找節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
- 查找完成后,找到的節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)不是待查找節(jié)點(diǎn),就是分值23應(yīng)該插入的位置
每次搜索操作,不管是找到還是沒(méi)找到,返回的節(jié)點(diǎn)都是目標(biāo)位置的前一個(gè)節(jié)點(diǎn),所以,需要判斷返回節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)的分值與給定分值的關(guān)系從而得知要查找的節(jié)點(diǎn)是否在跳表中。當(dāng)然也可以獲得要插入的位置,比如查找分值為21的節(jié)點(diǎn),返回的同樣是指向20的節(jié)點(diǎn),該節(jié)點(diǎn)的下一個(gè)位置就是分值為21應(yīng)該插入的位置
插入操作
插入操作實(shí)際上就是執(zhí)行了一遍搜索功能,由于插入一個(gè)節(jié)點(diǎn),會(huì)破壞某些節(jié)點(diǎn)的next數(shù)組。所以需要在搜索過(guò)程中記錄每一層的前一個(gè)節(jié)點(diǎn),以插入22為例,每一層的前一個(gè)節(jié)點(diǎn)分別是7,20,20三個(gè)節(jié)點(diǎn)(實(shí)際上是兩個(gè))
//t_zset.c /* 在跳表中插入節(jié)點(diǎn),值為score數(shù)據(jù)為obj */ /* 因?yàn)椴迦牍?jié)點(diǎn)會(huì)破壞原跳表的結(jié)構(gòu),所以需要先找到會(huì)被破壞的那些節(jié)點(diǎn) * 被破壞的節(jié)點(diǎn)是每一層插入位置的前一個(gè)節(jié)點(diǎn),因?yàn)樗膎ext數(shù)組需要更改 */ zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {/* update保存每一層插入位置的前一個(gè)節(jié)點(diǎn) */zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;unsigned int rank[ZSKIPLIST_MAXLEVEL];int i, level;serverAssert(!isnan(score));x = zsl->header;/* 尋找每一層插入位置的前一個(gè)節(jié)點(diǎn) */for (i = zsl->level-1; i >= 0; i--) {rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];/* 實(shí)際上就是跳表的查找規(guī)則* 如果當(dāng)前層上的next指針指向的節(jié)點(diǎn)分值大于要查找的分值,則在同層移動(dòng)* 如果當(dāng)前層上的next指針指向的節(jié)點(diǎn)分值小于要查找的分值,則降層,不移動(dòng) *//* 而如果要查找每一層插入位置的上一個(gè)節(jié)點(diǎn),那么降層時(shí)的節(jié)點(diǎn)就要要找的節(jié)點(diǎn) */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;/* 同層移動(dòng) */x = x->level[i].forward;}/* 如果一旦降層,當(dāng)前節(jié)點(diǎn)就是要查找的節(jié)點(diǎn) */update[i] = x;}/* 為新節(jié)點(diǎn)隨機(jī)生成一個(gè)層數(shù) */level = zslRandomLevel();/* 如果隨機(jī)出的層數(shù)大于跳表總層數(shù),那么將跳表擴(kuò)層 */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;}/* 創(chuàng)建插入的新節(jié)點(diǎn) */x = zslCreateNode(level,score,obj);/* 因?yàn)樾鹿?jié)點(diǎn)只存在[0 : level]層,所以對(duì)于高于level層的那些節(jié)點(diǎn)沒(méi)有影響 */for (i = 0; i < level; i++) {/* 每一層都相當(dāng)于鏈表插入 */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]);update[i]->level[i].span = (rank[0] - rank[i]) + 1;}/* 高層節(jié)點(diǎn)的跨度加一 */for (i = level; i < zsl->level; i++) {update[i]->level[i].span++;}/* 設(shè)置新節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)指針 */x->backward = (update[0] == zsl->header) ? NULL : update[0];/* 如果插入位置是跳表末尾,更新表尾節(jié)點(diǎn) */if (x->level[0].forward)x->level[0].forward->backward = x;elsezsl->tail = x;/* 節(jié)點(diǎn)個(gè)數(shù)增加 */zsl->length++;return x; }刪除操作
刪除操作首先在跳表中查找需要?jiǎng)h除的節(jié)點(diǎn),如果找到,則將其刪除。需要注意,刪除和插入一樣,都會(huì)破壞某些節(jié)點(diǎn)的next指針,所以需要更新
zslDelete函數(shù)用于找到匹配的節(jié)點(diǎn),zslDeleteNode函數(shù)用于將節(jié)點(diǎn)從跳表中刪除
//t_zset.c /* 從跳表中刪除節(jié)點(diǎn) */ void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {int i;/* 對(duì)于每一層,改變其next數(shù)組和跨度 */for (i = 0; i < zsl->level; i++) {/* 如果當(dāng)前節(jié)點(diǎn)的當(dāng)前層的next節(jié)點(diǎn)是要?jiǎng)h除的節(jié)點(diǎn),改變其next指針和跨度 */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;}}/* 刪除節(jié)點(diǎn)后面的后繼節(jié)點(diǎn)的前驅(qū)指針也需要改變 */if (x->level[0].forward) {x->level[0].forward->backward = x->backward;} else {zsl->tail = x->backward;}/* 如果刪除的是最高層節(jié)點(diǎn),同時(shí)刪除后最高層為空,就將跳表層數(shù)降低 */while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)zsl->level--;/* 節(jié)點(diǎn)個(gè)數(shù)減少 */zsl->length--; }/* 刪除分值score和數(shù)據(jù)obj匹配的節(jié)點(diǎn) */ int zslDelete(zskiplist *zsl, double score, robj *obj) {zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;int i;x = zsl->header;/* 尋找待刪除節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn),和插入操作的查找相同 */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;}/* 判斷是否存在要?jiǎng)h除的節(jié)點(diǎn),x是插入位置前的節(jié)點(diǎn),那么它的next指針就是需要?jiǎng)h除的節(jié)點(diǎn) */x = x->level[0].forward;if (x && score == x->score && equalStringObjects(x->obj,obj)) {/* 如果是,則調(diào)用刪除節(jié)點(diǎn)操作 */zslDeleteNode(zsl, x, update);zslFreeNode(x);return 1;}return 0; /* not found */ }計(jì)算某個(gè)節(jié)點(diǎn)的排名
Redis跳表可以計(jì)算某個(gè)數(shù)據(jù)在跳表中的排名,由zslGetRank函數(shù)完成,函數(shù)仍然使用查找方法
//t_zset.c /* 計(jì)算數(shù)據(jù)o在跳表中的排名 */ 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))) {/* 跨度代表當(dāng)前節(jié)點(diǎn)到下一個(gè)節(jié)點(diǎn)跳過(guò)了幾個(gè)節(jié)點(diǎn),所以排名需要增加跨度個(gè) */rank += x->level[i].span;x = x->level[i].forward;}/* 當(dāng)降層后,說(shuō)明當(dāng)前層的下一個(gè)節(jié)點(diǎn)的值已經(jīng)大于score了,需要降低一層繼續(xù)查找* 當(dāng)然也有可能已經(jīng)找到,所以需要判斷是否匹配 */if (x->obj && equalStringObjects(x->obj,o)) {return rank;}}return 0; }對(duì)象系統(tǒng)中的跳表
跳表是作為有序集合的底層實(shí)現(xiàn)存在的,在object.c文件中,可以看到創(chuàng)建有序集合時(shí)將編碼設(shè)置為跳表
//object.c /* 創(chuàng)建有序集合對(duì)象,底層使用跳表實(shí)現(xiàn) */ robj *createZsetObject(void) {/* 申請(qǐng)有序集合內(nèi)存 */zset *zs = zmalloc(sizeof(*zs));robj *o;/* 為有序集合創(chuàng)建字典 */zs->dict = dictCreate(&zsetDictType,NULL);/* 創(chuàng)建有序集合中的跳表 */zs->zsl = zslCreate();o = createObject(OBJ_ZSET,zs);/* 設(shè)置編碼格式為跳表 */o->encoding = OBJ_ENCODING_SKIPLIST;return o; }小結(jié)
跳表是較平衡的數(shù)據(jù)結(jié)構(gòu),實(shí)現(xiàn)簡(jiǎn)單,插入刪除等都是建立在搜索上的。可以發(fā)現(xiàn),搜索是先嘗試在高層上移動(dòng),因?yàn)橐苿?dòng)的跨度大,可以快速的達(dá)到目的地,當(dāng)不滿足在高層移動(dòng)的條件時(shí),再降層移動(dòng),直到降到最低層。
總結(jié)
以上是生活随笔為你收集整理的Redis源码剖析(十一)跳表的全部?jī)?nèi)容,希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇: Redis源码剖析(十)简单动态字符串s
- 下一篇: Redis源码剖析(十二)有序集合跳表实