終於來到記憶體儲存引擎這一段的尾聲了。今天要碰 Redis 裡很有代表性的資料結構:Sorted Set (有序集合,簡稱 ZSet)。
ZSet 每個成員(Member)都會綁一個浮點數(Score),集合裡的成員是唯一的,但分數可以重複,而且全部都會按分數從小到大排好。
為了實作這個,我要親手刻一個跳躍表(Skip List)。
要在記憶體中維護一個有序集合,還要兼顧插入、刪除和區間查詢,大概會想到幾種做法:
在 Redis 的 ZSet 裡,我要同時處理兩種高頻操作:
ZSCORE),預期要 $O(1)$。ZRANGE),預期要 $O(\log N + M)$($M$ 為回傳元素數)。單靠跳躍表查 score 要 $O(\log N)$。所以這邊直接組合兩種結構:
dict map[string]float64:存 member 到 score 的對應,拿來做 $O(1)$ 分數查詢。zskipList:存依 score 排序的跳躍表,用來做範圍查詢。type SortedSet struct {
dict map[string]float64
zsl *zskipList
}
我在 code/db/zset.go 裡手刻了跳躍表與 ZSet。
這個 skiplist 的 randomLevel() 我看了 Redis 原始碼好幾遍才搞懂,為什麼要用概率決定層高而不是固定高度。其實我第一次寫 insert() 的指標更新時還不小心寫出無窮迴圈,找了好久才發現 update 陣列沒接好。
每個節點 zskipNode 會隨機分配一個層數(Level)。插入時,從最高層向右向下找插入點,並記錄每層的前驅節點(用於更新指標):
func (zsl *zskipList) insert(score float64, member string) *zskipNode {
var update [maxLevel]*zskipNode
curr := zsl.header
// 從最高層向下尋找合適的插入位置
for i := zsl.level - 1; i >= 0; i-- {
for curr.level[i].forward != nil &&
(curr.level[i].forward.score < score ||
(curr.level[i].forward.score == score && curr.level[i].forward.member < member)) {
curr = curr.level[i].forward
}
update[i] = curr
}
// 隨機決定該新節點的層數
level := zsl.randomLevel()
// ... 調整跳躍表最大高度 ...
node := newZskipNode(level, score, member)
// 更新指標鏈結
for i := 0; i < level; i++ {
node.level[i].forward = update[i].level[i].forward
update[i].level[i].forward = node
}
// ... 更新 backward 與 tail 指標 ...
zsl.length++
return node
}
ZADD當呼叫 ZADD 時,我們首先查詢 dict 看該成員是否已經存在:
dict,並插入跳躍表。dict 中的分數。這是為了確保跳躍表的有序性不被打亂。func (db *DB) ZAdd(key string, score float64, member string) (int, error) {
db.mu.Lock()
defer db.mu.Unlock()
zs, err := db.getOrInitZSet(key)
if err != nil { return 0, err }
oldScore, exists := zs.dict[member]
if exists {
if oldScore == score { return 0, nil } // 分數未變
// 分數變了:從跳躍表移除舊的,插入新的
zs.zsl.remove(oldScore, member)
zs.zsl.insert(score, member)
zs.dict[member] = score
return 0, nil // 更新操作
}
// 全新插入
zs.zsl.insert(score, member)
zs.dict[member] = score
return 1, nil
}
ZRANGEZRANGE 的實作與 List 的 LRANGE 相似,透過將負數索引修正,定位到跳躍表中最底層(Level 0)的對應節點,然後順著 forward 指標向後遍歷收集元素即可:
func (db *DB) ZRange(key string, start, stop int) ([]ZSetMember, error) {
// ... 鎖定、存在性檢查與索引修正 ...
res := make([]ZSetMember, 0, stop-start+1)
curr := zs.zsl.header.level[0].forward
// 1. 移動到 start 位置
for i := 0; i < start && curr != nil; i++ {
curr = curr.level[0].forward
}
// 2. 依序收集
for i := start; i <= stop && curr != nil; i++ {
res = append(res, ZSetMember{
Member: curr.member,
Score: curr.score,
})
curr = curr.level[0].forward
}
return res, nil
}
到這裡,String、List、Hash、Set、Sorted Set 和 TTL 都先做完一版了。
接著跑一次測試,確認這些資料結構沒有互相踩到:
cd code
go test -v ./...
測試結果:
? redis-clone [no test files]
=== RUN TestDBBasic
--- PASS: TestDBBasic (0.00s)
=== RUN TestDBExpiration
--- PASS: TestDBExpiration (0.10s)
=== RUN TestList
--- PASS: TestList (0.00s)
=== RUN TestHash
--- PASS: TestHash (0.00s)
=== RUN TestSet
--- PASS: TestSet (0.00s)
=== RUN TestSortedSet
--- PASS: TestSortedSet (0.00s)
PASS
ok redis-clone/db 0.600s
...
PASS
ok redis-clone/resp 0.231s
測試都有過,至少目前這版資料結構的基本行為和鎖保護沒有明顯問題。
先把 Server 跑起來(順便跑個測試確認都 PASS):
# 跑測試
go test -v ./code/db/...
# 預期輸出:PASS
# 啟動 Server
go run ./code/main.go
接著開另一個終端機,用 printf + nc 送 RESP 命令來測試我們的 ZSet:
# ZADD scores 100 alice
printf "*4\r\n\$4\r\nZADD\r\n\$6\r\nscores\r\n\$3\r\n100\r\n\$5\r\nalice\r\n" | nc localhost 6379
# 預期回覆::1
# ZADD scores 90 bob 120 carol
printf "*6\r\n\$4\r\nZADD\r\n\$6\r\nscores\r\n\$2\r\n90\r\n\$3\r\nbob\r\n\$3\r\n120\r\n\$5\r\ncarol\r\n" | nc localhost 6379
# 預期回覆::2
# ZSCORE scores alice
printf "*3\r\n\$6\r\nZSCORE\r\n\$6\r\nscores\r\n\$5\r\nalice\r\n" | nc localhost 6379
# 預期回覆:$3\r\n100\r\n
# ZRANGE scores 0 -1
printf "*4\r\n\$6\r\nZRANGE\r\n\$6\r\nscores\r\n\$1\r\n0\r\n\$2\r\n-1\r\n" | nc localhost 6379
# 預期回覆:包含 bob, alice, carol 及它們分數的 Array 回覆
今天算是把跳躍表硬刻完了。Sorted Set 做起來後,記憶體引擎這個階段也差不多可以收尾。
明天開始進入命令處理器與事件迴圈,終於要把 TCP Server 和 DB 串在一起了,到時候見!