Go 語言下的 Redis 跳表設(shè)計與實現(xiàn)
在現(xiàn)代高性能數(shù)據(jù)庫和緩存系統(tǒng)中，跳表（Skip List）作為一種高效的有序數(shù)據(jù)結(jié)構(gòu)，被廣泛應(yīng)用于快速查找、插入和刪除操作。Redis 是一個開源的鍵值對存儲系統(tǒng)，它支持多種數(shù)據(jù)類型，并以其出色的性能而聞名。其中，Redis 使用了跳表來實現(xiàn)有序集合（Sorted Set），以保證其高效的數(shù)據(jù)處理能力。
本文將詳細介紹如何使用 Go 語言從零開始實現(xiàn)一個類似于 Redis 的跳表。我們將探討跳表的基本原理、設(shè)計思路以及具體的實現(xiàn)方法。通過本篇文章的學(xué)習(xí)，你不僅能夠了解跳表的工作機制，還能夠在實際項目中應(yīng)用這一強大的數(shù)據(jù)結(jié)構(gòu)。
定義基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)
redis中跳表通過score標(biāo)識元素的大小，通過redis obj維護節(jié)點的信息，與此同時為了保證查詢的高效，它會為每個節(jié)點維護一份隨機高度的索引記錄當(dāng)前節(jié)點的某個前驅(qū)節(jié)點：
對應(yīng)我們給出節(jié)點的代碼實現(xiàn)：
/*
*
跳表節(jié)點的定義
*/
type zskiplistNode struct {
//記錄元素的redis指針
obj *robj
//記錄當(dāng)前元素的數(shù)值，代表當(dāng)前元素的優(yōu)先級
score float64
//指向當(dāng)前元素的前驅(qū)節(jié)點，即小于當(dāng)前節(jié)點的元素
backward *zskiplistNode
//用一個zskiplistLevel數(shù)組維護本屆點各層索引信息
level []zskiplistLevel
}zskiplistLevel的代碼實現(xiàn)比較簡單，通過forward 記錄本層索引的前驅(qū)節(jié)點，并用span維護當(dāng)前節(jié)點需要跨幾步才能走到該前驅(qū)節(jié)點：
type zskiplistLevel struct {
//記錄本層索引的前驅(qū)節(jié)點的指針
forward *zskiplistNode
//標(biāo)識節(jié)點的本層索引需要跨幾步才能到達該節(jié)點
span int64
}通過上述概念構(gòu)成無數(shù)個節(jié)點即稱為跳表，如下圖所示，各個節(jié)點都用一個level數(shù)組記錄本層索引到前驅(qū)節(jié)點的地址和跨度，而跳表也用一個header和tail指針維護跳表的頭尾節(jié)點：
對應(yīng)的跳表結(jié)構(gòu)體的代碼如下所示：
type zskiplist struct {
//指向跳表的頭節(jié)點
header *zskiplistNode
//指向跳表的尾節(jié)點
tail *zskiplistNode
//維護跳表的長度
length int64
//維護跳表當(dāng)前索引的最高高度
level int
}實現(xiàn)初始化方法
對應(yīng)的我們也給出跳表的初始化代碼，大體邏輯是初始化跳表之后，初始化一個全空的索引和維護跳表的各種初始化信息，對應(yīng)的筆者也對此代碼做了詳盡的注釋，讀者可自行參閱：
func zslCreate() *zskiplist {
var j int
//初始化跳表結(jié)構(gòu)體
zsl := new(zskiplist)
//索引默認高度為1
zsl.level = 1
//跳表元素初始化為0
zsl.length = 0
//初始化一個頭節(jié)點socre為0，元素為空
zsl.header = zslCreateNode(ZSKIPLIST_MAXLEVEL, 0, nil)
/**
基于跳表最大高度32初始化頭節(jié)點的索引，
使得前驅(qū)指針指向null 跨度也設(shè)置為0
*/
for j = 0; j < ZSKIPLIST_MAXLEVEL; j++ {
zsl.header.level[j].forward = nil
zsl.header.level[j].span = 0
}
//頭節(jié)點的前驅(qū)節(jié)點指向null，代表頭節(jié)點之前沒有任何元素
zsl.header.backward = nil
//初始化尾節(jié)點
zsl.tail = nil
return zsl
}跳表插入操作
插入新節(jié)點時，本質(zhì)上就是通過各層索引找到小于插入節(jié)點x的score的最大值，并記錄到update數(shù)組中，同時將頭節(jié)點跨到update數(shù)組元素的跨度值記錄到rank數(shù)組中，如下圖所示，假如我們插入節(jié)點1.5，那么對應(yīng)各層索引的在update和rank兩個數(shù)組中維護的信息是：
- level2級中update記錄header節(jié)點，所以跨度為0。
- level1級中update記錄的是節(jié)點1，跨度為1。
然后基于此信息將x插入：
對應(yīng)的代碼和上述圖解邏輯一致，對應(yīng)的實現(xiàn)細節(jié)筆者都做好了標(biāo)注：
func zslInsert(zsl *zskiplist, score float64, obj *robj) *zskiplistNode {
//創(chuàng)建一個update數(shù)組，記錄插入節(jié)點每層索引中小于該score的最大值
update := make([]*zskiplistNode, ZSKIPLIST_MAXLEVEL)
//記錄各層索引走到小于score最大節(jié)點的跨區(qū)
rank := make([]int64, ZSKIPLIST_MAXLEVEL)
//x指向跳表走節(jié)點
x := zsl.header
var i int
//從跳表當(dāng)前最高層索引開始，查找每層小于當(dāng)前score的節(jié)點的最大值節(jié)點
for i = zsl.level - 1; i >= 0; i-- {
//如果當(dāng)前索引是最高層索引，那么rank從0開始算
if i == zsl.level-1 {
rank[i] = 0
} else { //反之本層索引直接從上一層的跨度開始往后查找
rank[i] = rank[i+1]
}
/**
如果前驅(qū)節(jié)點不為空，且符合以下條件，則指針前移：
1. 節(jié)點小于當(dāng)前插入節(jié)點的score
2. 節(jié)點score一致，且元素值小于或者等于當(dāng)前score
*/
if x.level[i].forward != nil &&
(x.level[i].forward.score < score || (x.level[i].forward.score == score && x.level[i].forward.obj.String() < obj.String())) {
//記錄本層索引前移跨度
rank[i] += x.level[i].span
//索引指針先前移動
x = x.level[i].forward
}
//記錄本層小于當(dāng)前score的最大節(jié)點
update[i] = x
}
//隨機生成新插入節(jié)點的索引高度
level := zslRandomLevel()
/**
如果大于當(dāng)前索引高度，則進行初始化，將這些高層索引的update數(shù)組都指向header節(jié)點，跨度設(shè)置為跳表中的元素數(shù)
意為這些高層索引小于插入節(jié)點的最大值就是header
*/
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
}
//基于入?yún)⑸梢粋€節(jié)點
x = zslCreateNode(level, score, obj)
//從底層到當(dāng)前最高層索引處理節(jié)點關(guān)系
for i = 0; i < level; i++ {
//將小于當(dāng)前節(jié)點的最大節(jié)點的forward指向插入節(jié)點x，同時x指向這個節(jié)點的前向節(jié)點
x.level[i].forward = update[i].level[i].forward
update[i].level[i].forward = x
//維護x和update所指向節(jié)點之間的跨度信息
x.level[i].span = update[i].level[i].span - (rank[0] - rank[i])
update[i].level[i].span = rank[0] - rank[i] + 1
}
/**
考慮到當(dāng)前插入節(jié)點生成的level小于當(dāng)前跳表最高level的情況
該邏輯會將這些區(qū)間的update索引中的元素到其前方節(jié)點的跨度＋1，即代表這些層級索引雖然沒有指向x節(jié)點，
但因為x節(jié)點插入的緣故跨度要加1
*/
for i = level; i < zsl.level; i++ {
update[i].level[i].span++
}
//如果1級索引是header，則x后繼節(jié)點不指向該節(jié)點，反之指向
if update[0] == zsl.header {
x.backward = nil
} else {
x.backward = update[0]
}
//如果x前向節(jié)點不為空，則讓前向節(jié)點指向x
if x.level[0].forward != nil {
x.level[0].forward.backward = x
} else {//反之說明x是尾節(jié)點，tail指針指向它
zsl.tail = x
}
//維護跳表長度信息
zsl.length++
return x
}跳表查詢操作
有了插入操作的基礎(chǔ)后，查詢操作實現(xiàn)也比較容易了，即從頭節(jié)點的最高索引開始不斷向前找，如果沒有則往下一級索引前向找，找到后返回經(jīng)過的跨度即可。
如下圖，我們希望查找元素2，直接從頭節(jié)點的2級索引開始看，就是元素2比對一致，返回跨度2，即跨2步就能到達：
對應(yīng)代碼如下，和筆者說明一致，這里筆者也做了詳盡的標(biāo)注提供參考：
func zslGetRank(zsl *zskiplist, score float64, obj *robj) int64 {
var rank int64
//從索引最高節(jié)點開始進行查找
x := zsl.header
for i := zsl.level - 1; i >= 0; i-- {
//如果前向節(jié)點不為空且score小于查找節(jié)點，或者score相等，但是元素字符序比值小于或者等于則前移，同時用rank記錄跨度
for x.level[i].forward != nil &&
(x.level[i].forward.score < score || (x.level[i].forward.score == score && x.level[i].forward.obj.String() <= obj.String())) {
rank += x.level[i].span
x = x.level[i].forward
}
//上述循環(huán)結(jié)束，比對一直，則返回經(jīng)過的跨度
if x.obj != nil && x.obj.String() == obj.String() {
return rank
}
}
return 0
}跳表刪除操作
刪除操作本質(zhì)上也是找到要刪除節(jié)點索引的前后節(jié)點，然后將這些節(jié)點關(guān)聯(lián)，并修改其之間跨度，如下圖我們要刪除1.5節(jié)點，對應(yīng)各層查找結(jié)果為：
- 3級索引找到頭節(jié)點，因為前方不是1.5的節(jié)點索引，直接跨度減1即。
- 2級索引找到頭節(jié)點，前方就是1.5的索引，刪除掉后跨度改為header索引到1.5+1.5到前向節(jié)點跨度減去1，這里的減去1代表刪除了節(jié)點1.5的跨步。
- 1級索引同2級索引，不多做贅述。
對應(yīng)的代碼示例如下，整體邏輯和筆者描述基本一致，先通過update找到刪除節(jié)點x的前一個元素，然后調(diào)用zslDeleteNode進行刪除：
func zslDelete(zsl *zskiplist, score float64, obj *robj) int64 {
update := make([]*zskiplistNode, ZSKIPLIST_MAXLEVEL)
//找到每層索引要刪除節(jié)點的前一個節(jié)點
x := zsl.header
for i := zsl.level - 1; i >= 0; i-- {
for x.level[i].forward != nil &&
(x.level[i].forward.score < score || (x.level[i].forward.score == score && x.level[i].forward.obj.String() < obj.String())) {
x = x.level[i].forward
}
update[i] = x
}
//查看1級索引前面是否就是要刪除的節(jié)點，如果是則直接調(diào)用zslDeleteNode刪除節(jié)點，并斷掉前后節(jié)點關(guān)系
x = x.level[0].forward
if x != nil && x.obj.String() == obj.String() {
zslDeleteNode(zsl, x, update)
return 1
}
return 0
}對應(yīng)zslDeleteNode細節(jié)就如筆者上圖所講解的步驟，讀者可參考注釋進行閱讀：
func zslDeleteNode(zsl *zskiplist, x *zskiplistNode, update []*zskiplistNode) {
var i int
for i = 0; i < zsl.level; i++ {
/*
如果索引前方就是刪除節(jié)點，當(dāng)前節(jié)點span為：
當(dāng)前節(jié)點到x +x到x前向節(jié)點 -1
*/
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 {
//反之說明該節(jié)點前方不是x的索引，直接減去x的跨步1即
update[i].level[i].span -= 1
}
}
//維護刪除后的節(jié)點前后關(guān)系
if x.level[0].forward != nil {
x.level[0].forward.backward = x.backward
} else {
zsl.tail = x.backward
}
//將全空層的索引刪除
for zsl.level > 1 && zsl.header.level[zsl.level-1].forward == nil {
zsl.level--
}
//維護跳表節(jié)點信息
zsl.length--小結(jié)
通過本文的詳細講解，我們從零開始使用 Go 語言實現(xiàn)了一個類似于 Redis 的跳表。我們首先介紹了跳表的基本原理和設(shè)計思路，然后逐步實現(xiàn)了跳表的各種核心操作，包括插入、查找和刪除。最后，我們對跳表的性能進行了分析，并探討了其在 Redis 有序集合和其他場景中的應(yīng)用。
