Redis跳錶與有序集合實現
為了大家看整體原始碼方便，我將加上了完整註釋的程式碼傳到了我的github上供大家直接下載：
在Redis中提供了有序集合，從它的命令之多就可以知道它的功能是比較強大的，當然實現也是比較複雜的，ZSET底層使用跳錶SKIP LIST實現，在Java中也有跳錶主要用於實現併發訪問。
跳錶
雖然這不是我們的主題，但是你需要對跳錶這種常用資料結構有一定的認識，否則很難理解後面的排序輸出，範圍查詢等功能。
跳錶本質上是一種查詢結構，它解決的問題是能夠更加快速的查詢到想找的元素，相對與紅黑樹和其它平衡樹查詢與插入的邏輯，跳錶是非常好上手的。
藉助跳錶結構提出者William Pugh給的一張圖，可以生動形象表示跳錶的基本思想。
比如咱們要查詢數字16，當然沒有，但咱看看要查幾次才知道沒有呢。
先看第a行，這就是一個普通的list，單向連結串列，想查詢一個元素16，要沿著列表走6步，3 -> 6 -> 7 -> 9 -> 12 -> 17，才知道沒有。
到了第b行，我們加了一層，將相隔2步的元素提到上一層，查元素的時候，我們先從高層查起，只需要4步，6 -> 9 -> 17 -> 12，每次查詢跨的步子都大了，第一步就查了6，此時我們知道6左下層的元素不需要查了，每次跨的步子大了，查詢的次數自然也就少了。
每層的元素個數都是下一層的一半，每多一層元素減少50%，相當於二分查詢法，相比列表查詢元素的時間複雜度從O(n)
O(log n)。
跳錶能否提升查詢效能在於分層，過多的層會導致空間損失和插入效能損失，每一層能夠跨的元素越多越好，那如何把哪幾個元素提高一層能提供查詢效能呢？很難衡量，二層來說可以通過計算元素間距離來得到，但是三層四層呢，這一層的結果影響下一層的提層，這層分的好可能導致下一層分的不好，反之亦然。 而且根據固定位置分層會導致每次插入元素都可能導致各元素層高變化，代價很高。
所以在William Pugh使用的一種隨機層數策略，每一個元素插進來時，它的層數是隨機生成的，這是跳錶很重要的特性。那隨機的效能如何呢？在原論文中有一章節Analysis of expected search cost
查詢的過程比較簡單想必大家已經很清楚，作者用一段偽程式碼表示了插入的邏輯
-- 和lua語法註釋一樣
Insert(list, searchKey, newValue)
-- Redis中的程式碼實現以及變數命名都和此很像
-- update儲存的是各個層級上新插入元素位置的前一個位置
local update[1..MaxLevel]
x := list→header
-- 遍歷每一層直到找到新元素的位置，並記錄該位置的前一個元素
for i := list→level downto 1 do
while x→forward[i]→key < searchKey do
x := x→forward[i]
-- x→key < searchKey ≤ x→forward[i]→key
update[i] := x
x := x→forward[1]
-- 存在相同key（相同排序依據分數）則替換那個位置，不允許有相同分數的元素
if x→key = searchKey then x→value := newValue
else
lvl := randomLevel()
if lvl > list→level then
-- 如果出現新層級高於目前最高層級的情況
for i := list→level + 1 to lvl do
update[i] := list→header
list→level := lvl
x := makeNode(lvl, searchKey, value)
-- 把元素插進到每一層（它指向前節點的下一個節點，再將前節點改為指向它）
for i := 1 to level do
x→forward[i] := update[i]→forward[i]
update[i]→forward[i] := x
當然看一遍不可能理解的很透徹，但是大概有個概念，不要影響後續對有序集合的分析即可。
Redis中的zskiplist
大多數對跳躍表的實現都會根據場景進行修改，Redis根據要支撐的有序集合ZSET的特性，對跳躍表進行一下節點修改。
在Redis中用zskiplist和zskiplistNode分別表示跳錶和跳錶節點
/*
* 跳錶的具體節點
*/
typedef struct zskiplistNode {
// 實際元素資料對應字串，在存入跳錶前會被編碼為字串
// Redis還會將此ele作為key,分數儲存在字典中方便統計
sds ele;
// 排序依據, 允許多個同分數不同元素存在
double score;
// 後節點指標，Redis的跳錶第一層是一個雙向連結串列
struct zskiplistNode *backward;
// 表示一個節點共有多少層, 是一個柔性陣列，需要在建立節點時根據層高具體分配
struct zskiplistLevel {
// 前節點指標
struct zskiplistNode *forward;
// 該層一次元素跳躍一共跳過多少個第一層元素, 用於統計排名
unsigned int span;
} level[];
} zskiplistNode;
/*
* Redis使用的跳錶, 是有序集合zset的底層實現
*/
typedef struct zskiplist {
// 頭尾節點
struct zskiplistNode *header, *tail;
// 跳錶共有元素個數
unsigned long length;
// 跳錶目前最高的層數
int level;
} zskiplist;
根據結構體zskiplistNode可以較好的理解Redis中跳錶的實現，和標準跳躍表實現有幾個小的區別。
- 排序依據分數允許重複，相同分數根據元素資料ele字串自然排序，但元素值不可重複
- 第0層是一個雙向連結串列，和列表一樣，方便倒序取資料
- 增加了統計類屬性，方便排名與計數
Redis中的跳錶操作
對跳錶本身無非是增刪改查，我們就看一下插入即可，因為它包含了查詢，插入的邏輯和前面標準跳錶的虛擬碼幾乎一致，只是細節上有區分，這樣我們也可以對比下Redis跳錶和標準跳錶區別。
/*
* 將元素插入到跳錶中
*
* 引數列表
* 1. zsl: 跳錶結構體
* 2. score: 插入元素的分數
* 3. ele: 插入元素的實際資料
*
* 返回值
* 插入元素的對應節點
*/
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
// 和標準跳錶一樣使用update陣列記錄每層待插入位置所在前一個元素
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
// 記錄前置節點與第一個節點之間的跨度,即元素在列表中的排名-1
// 跨度指的都是跨過第0層多少個元素
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;
}
// 隨機產生一個層數，在1與MAXLEVEL之間，層數越高生成概率越低
level = zslRandomLevel();
if (level > zsl->level) {
// 如果產生的層數大於現有最高層數，則超出層數都需要初始化
for (i = zsl->level; i < level; i++) {
rank[i] = 0;
// 該元素作為這些層的第一個節點，前節點就是header
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;
// rank[0]是第0層的前置節點P1（也就是底層插入節點前面那個節點）與第一個節點的跨度
// rank[i]是第i層的前置節點P2（這一層裡在插入節點前面那個節點）與第一個節點的跨度
// 插入節點X與後置節點Y的跨度f(X,Y)可由以下公式計算
// 關鍵在於f(P1,0)-f(P2,0)+1等於新節點與P2的跨度，這是因為跨度呈梯子形向下延伸到最底層
// 記錄節點各層跨越元素情況span, 由層與層之間的跨越元素總和rank相減而得
x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
// 插入位置前一個節點的span在原基礎上加1即可(新節點在rank[0]的後一個位置)
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
// header是個起始
for (i = level; i < zsl->level; i++) {
update[i]->level[i].span++;
}
! // 第0層是雙向連結串列, 便於redis常支援逆序類查詢
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;
}
大家可以看到跳錶的元素定位、插入都還是比較繁瑣的，如果少量資料就使用跳錶是得不償失的。
Redis中的ZSET實現
在Redis中有序集合的實現，不完全是使用跳錶，在資料量少的情況下（128以下），Redis會使用壓縮連結串列ziplist來實現，當資料量超過閾值才會使用跳錶，ziplist相關的程式碼比較簡單，僅一筆帶過，接下來討論跳錶模式下的場景。
某些情況下，如獲取某個元素的分數、求集合並集等情況，需要元素值與其分數的對應關係，簡單的做法當然遍歷一下跳錶，找到這個元素node，自然得到它的分數。
但Redis為了提高效率，直接將元素資料ele和其分數score的對應關係存在了雜湊表中，便於快速查詢，比如ZSCORE命令的實現概要如下：
/*
* 獲取指定元素的分數
*/
int zsetScore(robj *zobj, sds member, double *score) {
if (!zobj || !member) return C_ERR;
// ziplist模式下直接找到該元素並設定分數結果
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
if (zzlFind(zobj->ptr, member, score) == NULL) return C_ERR;
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
// 根據元素資料ele直接找到分數
dictEntry *de = dictFind(zs->dict, member);
if (de == NULL) return C_ERR;
*score = *(double*)dictGetVal(de);
} else {
serverPanic("Unknown sorted set encoding");
}
return C_OK;
}
通過冗餘一個雜湊表，使得查詢元素分數非常方便。
通過ZSCORE命令可以理解到Redis有序集合的實現概要，通過最常用的ZRANGE命令則可以理解元素的查詢過程。
/*
* 獲取指定範圍的元素
*/
void zrangeGenericCommand(client *c, int reverse) {
robj *key = c->argv[1];
robj *zobj;
// 是否同時展示元素的分數
int withscores = 0;
// 從哪個位置到哪個位置，尾可以負數表示倒數第幾個
long start;
long end;
int llen;
int rangelen;
...獲取傳遞的引數並賦值給本地變數
// 沒有這個zset或者key對應元素型別不是zset
if ((zobj = lookupKeyReadOrReply(c,key,shared.emptymultibulk)) == NULL
|| checkType(c,zobj,OBJ_ZSET)) return;
llen = zsetLength(zobj);
if (start < 0) start = llen+start;
if (end < 0) end = llen+end;
// 轉了一圈以上了，就認為從頭開始
if (start < 0) start = 0;
// 嚴謹的index range check
if (start > end || start >= llen) {
addReply(c,shared.emptymultibulk);
return;
}
if (end >= llen) end = llen-1;
// 一個要輸出多少個元素
rangelen = (end-start)+1;
addReplyMultiBulkLen(c, withscores ? (rangelen*2) : rangelen);
// 在元素較少時，zset底層使用ziplist實現，之前已解析過ziplist，此場景可認為是普通連結串列
if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
unsigned char *zl = zobj->ptr;
unsigned char *eptr, *sptr;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
// 移動到指定下標位置，準備開始遍歷
if (reverse)
eptr = ziplistIndex(zl,-2-(2*start));
else
eptr = ziplistIndex(zl,2*start);
serverAssertWithInfo(c,zobj,eptr != NULL);
sptr = ziplistNext(zl,eptr);
// 一個個遍歷，共遍歷rangelen個元素輸出即可
while (rangelen--) {
...遍歷輸出
}
} else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
// 當元素到達一定數量才使用跳錶, 預設域值為OBJ_ZSET_MAX_ZIPLIST_ENTRIES=128
zset *zs = zobj->ptr;
zskiplist *zsl = zs->zsl;
zskiplistNode *ln;
sds ele;
if (reverse) {
ln = zsl->tail;
// start==0時就是從頭或尾開始查詢
if (start > 0)
ln = zslGetElementByRank(zsl,llen-start);
} else {
ln = zsl->header->level[0].forward;
// 根據跨度span計數來找到排名為start+1的節點
if (start > 0)
ln = zslGetElementByRank(zsl,start+1);
}
// 從起始位置開始輸出rangelen個節點
while(rangelen--) {
serverAssertWithInfo(c,zobj,ln != NULL);
ele = ln->ele;
addReplyBulkCBuffer(c,ele,sdslen(ele));
if (withscores)
addReplyDouble(c,ln->score);
ln = reverse ? ln->backward : ln->level[0].forward;
}
} else {
serverPanic("Unknown sorted set encoding");
}
}
看完Redis的有序集合實現，當時我也有個疑惑，為什麼不用平衡樹實現（也疑惑Redis的雜湊表在雜湊衝突時為什麼不用樹實現D–），以下我自己的理解。
- 跳錶實現起來簡單，這個很重要，也和Redis的宗旨符合，且效能相當
- 跳錶更適合範圍查詢
在實際環境中，使用ZSET完成排行榜模組是非常常見的，點贊量、閱讀數量、播放量等等，它可以多維度滿足排行需求且操作簡單。
好啦，講完，希望對你有所幫助。