Golang 实现 Redis(5): 用跳表实现SortedSet
- 2020 年 5 月 8 日
- 笔记
- 用 Golang 实现 Redis
本文是使用 golang 实现 redis 系列的第五篇， 将介绍如何使用跳表实现有序集合(SortedSet)的相关功能。
跳表(skiplist) 是 Redis 中 SortedSet 数据结构的底层实现, 跳表优秀的范围查找能力为ZRange和ZRangeByScore等命令提供了支持。
本文完整源代码在GithubHDT3213/godis
结构定义
实现ZRange命令最简单的数据结构是有序链表:
在有序链表上实现ZRange key start end命令需要进行end次查询, 即时间复杂度为 O(n)
跳表的优化思路是添加上层链表，上层链表中会跳过一些节点。如图所示:
在有两层的跳表中，搜索的时间复杂度降低为了O(n / 2)。以此类推在有 log2(n) 层的跳表中，搜索元素的时间复杂度为O(log n)。
了解数据结构之后，可以定义相关的类型了:
// 对外的元素抽象
type Element struct {
Member string
Score float64
}
type Node struct {
Element // 元素的名称和 score
backward *Node // 后向指针
level []*Level // 前向指针, level[0] 为最下层
}
// 节点中每一层的抽象
type Level struct {
forward *Node // 指向同层中的下一个节点
span int64 // 到 forward 跳过的节点数
}
// 跳表的定义
type skiplist struct {
header *Node
tail *Node
length int64
level int16
}
用一张图来表示一下:
查找节点
有了上文的描述查找节点的逻辑不难实现, 以 RangeByRank 的核心逻辑为例:
// 寻找排名为 rank 的节点, rank 从1开始
func (skiplist *skiplist) getByRank(rank int64)*Node {
var i int64 = 0
n := skiplist.header
// 从顶层向下查询
for level := skiplist.level - 1; level >= 0; level-- {
// 从当前层向前搜索
// 若当前层的下一个节点已经超过目标 (i+n.level[level].span > rank)，则结束当前层搜索进入下一层
for n.level[level].forward != nil && (i+n.level[level].span) <= rank {
i += n.level[level].span
n = n.level[level].forward
}
if i == rank {
return n
}
}
return nil
}
插入节点
插入节点的操作比较多，我们以注释的方式进行说明:
func (skiplist *skiplist)insert(member string, score float64)*Node {
// 寻找新节点的先驱节点，它们的 forward 将指向新节点
// 因为每层都有一个 forward 指针, 所以每层都会对应一个先驱节点
// 找到这些先驱节点并保存在 update 数组中
update := make([]*Node, maxLevel)
rank := make([]int64, maxLevel) // 保存各层先驱节点的排名，用于计算span
node := skiplist.header
for i := skiplist.level - 1; i >= 0; i-- { // 从上层向下寻找
// 初始化 rank
if i == skiplist.level - 1 {
rank[i] = 0
} else {
rank[i] = rank[i + 1]
}
if node.level[i] != nil {
// 遍历搜索
for node.level[i].forward != nil &&
(node.level[i].forward.Score < score ||
(node.level[i].forward.Score == score && node.level[i].forward.Member < member)) { // same score, different key
rank[i] += node.level[i].span
node = node.level[i].forward
}
}
update[i] = node
}
level := randomLevel() // 随机决定新节点的层数
// 可能需要创建新的层
if level > skiplist.level {
for i := skiplist.level; i < level; i++ {
rank[i] = 0
update[i] = skiplist.header
update[i].level[i].span = skiplist.length
}
skiplist.level = level
}
// 创建新节点并插入跳表
node = makeNode(level, score, member)
for i := int16(0); i < level; i++ {
// 新节点的 forward 指向先驱节点的 forward
node.level[i].forward = update[i].level[i].forward
// 先驱节点的 forward 指向新节点
update[i].level[i].forward = node
// 计算先驱节点和新节点的 span
node.level[i].span = update[i].level[i].span - (rank[0] - rank[i])
update[i].level[i].span = (rank[0] - rank[i]) + 1
}
// 新节点可能不会包含所有层
// 对于没有层，先驱节点的 span 会加1 (后面插入了新节点导致span+1)
for i := level; i < skiplist.level; i++ {
update[i].level[i].span++
}
// 更新后向指针
if update[0] == skiplist.header {
node.backward = nil
} else {
node.backward = update[0]
}
if node.level[0].forward != nil {
node.level[0].forward.backward = node
} else {
skiplist.tail = node
}
skiplist.length++
return node
}
randomLevel 用于随机决定新节点包含的层数，随机结果出现2的概率是出现1的25%， 出现3的概率是出现2的25%:
func randomLevel() int16 {
level := int16(1)
for float32(rand.Int31()&0xFFFF) < (0.25 * 0xFFFF) {
level++
}
if level < maxLevel {
return level
}
return maxLevel
}
删除节点
删除节点的思路与插入节点基本一致:
// 删除操作可能一次删除多个节点
func (skiplist *skiplist) RemoveRangeByRank(start int64, stop int64)(removed []*Element) {
var i int64 = 0 // 当前指针的排名
update := make([]*Node, maxLevel)
removed = make([]*Element, 0)
// 从顶层向下寻找目标的先驱节点
node := skiplist.header
for level := skiplist.level - 1; level >= 0; level-- {
for node.level[level].forward != nil && (i+node.level[level].span) < start {
i += node.level[level].span
node = node.level[level].forward
}
update[level] = node
}
i++
node = node.level[0].forward // node 是目标范围内第一个节点
// 删除范围内的所有节点
for node != nil && i < stop {
next := node.level[0].forward
removedElement := node.Element
removed = append(removed, &removedElement)
skiplist.removeNode(node, update)
node = next
i++
}
return removed
}
接下来分析一下执行具体节点删除操作的removeNode函数:
// 传入目标节点和删除后的先驱节点
// 在批量删除时我们传入的 update 数组是相同的
func (skiplist *skiplist) removeNode(node *Node, update []*Node) {
for i := int16(0); i < skiplist.level; i++ {
// 如果先驱节点的forward指针指向了目标节点，则需要修改先驱的forward指针跳过要删除的目标节点
// 同时更新先驱的 span
if update[i].level[i].forward == node {
update[i].level[i].span += node.level[i].span - 1
update[i].level[i].forward = node.level[i].forward
} else {
update[i].level[i].span--
}
}
// 修改目标节点后继节点的backward指针
if node.level[0].forward != nil {
node.level[0].forward.backward = node.backward
} else {
skiplist.tail = node.backward
}
// 必要时删除空白的层
for skiplist.level > 1 && skiplist.header.level[skiplist.level-1].forward == nil {
skiplist.level--
}
skiplist.length--
}
