Redis5.0原始碼解析(四)----------跳躍表
基於Redis5.0
跳躍表(skiplist)是一種有序資料結構, 它通過在每個節點中維持多個指向其他節點的指標, 從而達到快速訪問節點的目的
跳躍表支援平均 O(log N)
最壞 O(N)
複雜度的節點查詢, 還可以通過順序性操作來批量處理節點。
在大部分情況下, 跳躍表的效率可以和平衡樹相媲美, 並且因為跳躍表的實現比平衡樹要來得更為簡單, 所以有不少程式都使用跳躍表來代替平衡樹。
Redis 使用跳躍表作為有序集合鍵的底層實現之一: 如果一個有序集合包含的元素數量比較多, 又或者有序集合中元素的成員(member)是比較長的字串時, Redis 就會使用跳躍表來作為有序集合鍵的底層實現
為什麼選擇跳躍表
從該有序表中搜索元素 < 23, 43, 59 > ,需要比較的次數分別為 < 2, 4, 6 >,總共比較的次數為 2 + 4 + 6 = 12 次。連結串列是有序的,但不能使用二分查詢。類似二叉搜尋樹,我們把一些節點提取出來,作為索引。得到如下結構:
這裡我們把 < 14, 34, 50, 72 > 提取出來作為一級索引,這樣搜尋的時候就可以減少比較次數了。我們還可以再從一級索引提取一些元素出來,作為二級索引,變成如下結構:
如果元素足夠多,這種索引結構就能體現出優勢來了
Redis跳躍表的實現
最左邊的是 zskiplist
header
:指向跳躍表的表頭節點。tail
:指向跳躍表的表尾節點。level
:記錄目前跳躍表內,層數最大的那個節點的層數(表頭節點的層數不計算在內)。length
:記錄跳躍表的長度,也即是,跳躍表目前包含節點的數量(表頭節點不計算在內)。
位於 zskiplist
結構右方的是四個 zskiplistNode
結構:
level
:節點中用 L1 、 L2 、 L3 等字樣標記節點的各個層, L1 代表第一層, L2 代表第二層,以此類推。每個層都帶有兩個屬性:前進指標和跨度。前進指標用於訪問位於表尾方向的其他節點,而跨度則記錄了前進指標所指向節點和當前節點的距離。在上面的圖片中,連線上帶有數字的箭頭就代表前進指標,而那個數字就是跨度。當程式從表頭向表尾進行遍歷時,訪問會沿著層的前進指標進行。backward
:節點中用 BW 字樣標記節點的後退指標,它指向位於當前節點的前一個節點。後退指標在程式從表尾向表頭遍歷時使用。score
:各個節點中的 1.0 、 2.0 和 3.0 是節點所儲存的分值。在跳躍表中,節點按各自所儲存的分值從小到大排列。sds
:各個節點中的 o1 、 o2 和 o3 是節點所儲存的sds物件。
跳躍表節點:
//server.h
/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
//儲存的sds
sds ele;
//分值
double score;
// 後退指標
struct zskiplistNode *backward;
// 層
struct zskiplistLevel {
// 前進指標
struct zskiplistNode *forward;
/**
* 跨度實際上是用來計算元素排名(rank)的,
* 在查詢某個節點的過程中,將沿途訪過的所有層的跨度累積起來,
* 得到的結果就是目標節點在跳躍表中的排位
*/
unsigned long span;
} level[];
} zskiplistNode;
-
層
跳躍表節點的level
陣列可以包含多個元素, 每個元素都包含一個指向其他節點的指標, 程式可以通過這些層來加快訪問其他節點的速度, 一般來說, 層的數量越多, 訪問其他節點的速度就越快。每次建立一個新跳躍表節點的時候, 程式都根據冪次定律 (power law,越大的數出現的概率越小) 隨機生成一個介於 1 和 32 之間的值作為
level
陣列的大小, 這個大小就是層的“高度”。 -
前進指標
每個層都有一個指向表尾方向的前進指標(level[i].forward
屬性), 用於從表頭向表尾方向訪問節點。 -
跨度
層的跨度(level[i].span 屬性
)用於記錄兩個節點之間的距離跨度是用來計算排位(rank)的: 在查詢某個節點的過程中, 將沿途訪問過的所有層的跨度累計起來, 得到的結果就是目標節點在跳躍表中的排位
-
後退指標
節點的後退指標(backward
屬性)用於從表尾向表頭方向訪問節點: 跟可以一次跳過多個節點的前進指標不同, 因為每個節點只有一個後退指標, 所以每次只能後退至前一個節點 -
分值和成員
節點的分值(score
屬性)是一個 double 型別的浮點數, 跳躍表中的所有節點都按分值從小到大來排序。節點的成員(
sds
屬性)是一個字串sds
跳躍表
//server.h
typedef struct zskiplist {
// 表頭節點和表尾節點
struct zskiplistNode *header, *tail;
// 表中節點的數量
unsigned long length;
// 表中層數最大的節點的層數
int level;
} zskiplist;
header
和tail
指標分別指向跳躍表的表頭和表尾節點, 通過這兩個指標, 程式定位表頭節點和表尾節點的複雜度為O(1)
length
屬性來記錄節點的數量, 程式可以在O(1)
複雜度內返回跳躍表的長度level
屬性則用於在O(1)
複雜度內獲取跳躍表中層高最大的那個節點的層數量, 注意表頭節點的層高並不計算在內
跳躍表API實現:
建立一個跳躍表節點:
//t_zset.c
/* Create a skiplist node with the specified number of levels.
* The SDS string 'ele' is referenced by the node after the call. */
zskiplistNode *zslCreateNode(int level, double score, sds ele) {
//給Node以及level陣列分配記憶體
zskiplistNode *zn =
zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
zn->score = score;
zn->ele = ele;
return zn;
}
建立一個跳躍表:
/* Create a new skiplist. */
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
zsl = zmalloc(sizeof(*zsl));
zsl->level = 1;
zsl->length = 0;
zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
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;
return zsl;
}
在跳躍表中插入一個節點:
/* Insert a new node in the skiplist. Assumes the element does not already
* exist (up to the caller to enforce that). The skiplist takes ownership
* of the passed SDS string 'ele'. */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score));
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* store rank that is crossed to reach the insert position */
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
while (x->level[i].forward &&
(x->level[i].forward->score < score ||
(x->level[i].forward->score == score &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
/* we assume the element is not already inside, since we allow duplicated
* scores, reinserting the same element should never happen since the
* caller of zslInsert() should test in the hash table if the element is
* already inside or not. */
level = zslRandomLevel();
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;
}
x = zslCreateNode(level,score,ele);
for (i = 0; i < level; i++) {
x->level[i].forward = update[i]->level[i].forward;
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++) {
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;
}
在跳躍表中刪除一個節點:
/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
for (i = 0; i < zsl->level; i++) {
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->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--;
}