golang 实现线程安全的跳转表 （skip list）

redisredissorted setredisskip list

基础

skip list

只是列举了部分我觉得比较好的。如果你看到了更牛逼的文章可以在下面评论出来。
如果你看完了还是不理解可以找下别的文章再理解下。

go 代码实现

跳转表结构

直接看原始论文中的配图吧。

指针数据e61725

// elementNode 数组指针，指向元素
type elementNode struct {
    next []*Element
}

// Element 跳转表数据结构
type Element struct {
    elementNode
    key   float64 // 用以排序和判断大小的关键字
    value interface{} // 定义元素，附加
}

skip list

指针数组skip listmaxLevelemaxLevellevellevel

type SkipList struct {
    elementNode
    maxLevel       int            // 最大深度
    length         int            // 长度,跳表中
    randSource     rand.Source    // 动态调节跳转表的长度
    probability    float64        // 概率
    probTable      []float64      // 存储位置，对应key
    mutex          sync.RWMutex   // 保证线程安全
    prevNodesCache []*elementNode // 缓存
}

含义解释：

elementNodeprobTablemutexprevNodesCachelengthmaxLevel

基本方法

const (
    // DefaultMaxLevel 默认skip list最大深度
    DefaultMaxLevel int = 18
    // DefaultProbability 默认的概率
    DefaultProbability float64 = 1 / math.E
)

// Key 获取key的值
func (e *Element) Key() float64 {
    return e.key
}

// Value 获取key的值
func (e *Element) Value() interface{} {
    return e.value
}


type SkipList struct {
    elementNode
    maxLevel       int            // 最大深度
    length         int            // 长度
    randSource     rand.Source    // 随机数种子，动态调节跳转表元素的层数
    probability    float64        // 概率
    probTable      []float64      // 存储位置，对应key
    mutex          sync.RWMutex   // 保证线程安全
    prevNodesCache []*elementNode // 缓存
}

// NewSkipList 新建跳转表
func NewSkipList() *SkipList {
    return NewWithMaxLevel(DefaultMaxLevel)
}

// ProbabilityTable 初始化 Probability Table
func ProbabilityTable(probability float64, maxLevel int) (table []float64) {
    for i := 1; i <= maxLevel; i++ {
        prob := math.Pow(probability, float64(i-1))
        table = append(table, prob)
    }
    return table
}

// NewWithMaxLevel 自定义maxLevel新建跳转表
func NewWithMaxLevel(maxLevel int) *SkipList {
    if maxLevel < 1 || maxLevel > DefaultMaxLevel {
        panic("invalid maxlevel")
    }

    return &SkipList{
        elementNode:    elementNode{next: make([]*Element, maxLevel)},
        prevNodesCache: make([]*elementNode, maxLevel),
        maxLevel:       maxLevel,
        randSource:     rand.New(rand.NewSource(time.Now().UnixNano())),
        probability:    DefaultProbability,
        probTable:      ProbabilityTable(DefaultProbability, maxLevel),
    }
}

// 随机计算最接近的
func (list *SkipList) randLevel() (level int) {
    r := float64(list.randSource.Int63()) / (1 << 63)
    level = 1
    for level < list.maxLevel && r < list.probTable[level] {
        level++ // 级别追加
    }

    return level
}

// SetProbability 设置新的概率,刷新概率表
func (list *SkipList) SetProbability(newProbability float64) {
    list.probability = newProbability
    list.probTable = ProbabilityTable(newProbability, list.maxLevel)
}

方法解释：

NewWithMaxLevelelementNodeprevNodesCacherandLevel

核心方法

跳表中元素的查找效率和AVL树差不多，那这个效率是怎么实现这么高的我们就在这里说一下。

首先我们看下论文中出现的查询的一个说明图，看最上面的那个图就好

12

6669912

其实用语言来表述不是很准确，那就直接看代码

// Get 获取key对应的值
func (list *SkipList) Get(key float64) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var prev *elementNode = &list.elementNode // 保存前置结点
    var next *Element

    for i := list.maxLevel - 1; i >= 0; i-- {
        next = prev.next[i] // 循环跳到下一个
        for next != nil && key > next.key {
            prev = &next.elementNode
            next = next.next[i]
        }
    }

    if next != nil && next.key == key { // 找到
        return next
    }

    return nil // 没有找到
}

自己理解的时候可以边在自己脑袋里面一步一步的执行这些代码，然后再结合图中的结构就很快就理解了。

接下来看跳表的插入和删除了。

func (list *SkipList) getPrevElementNodes(key float64) []*elementNode {
    var prev *elementNode = &list.elementNode // 保存前置结点
    var next *Element
    prevs := list.prevNodesCache // 缓冲集合
    for i := list.maxLevel - 1; i >= 0; i-- {
        next = prev.next[i] // 循环跳到下一个
        for next != nil && key > next.key {
            prev = &next.elementNode
            next = next.next[i]
        }
        prevs[i] = prev
    }
    return prevs
}

// Set 存储新的值
func (list *SkipList) Set(key float64, value interface{}) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var element *Element
    prevs := list.getPrevElementNodes(key)
    if element = prevs[0].next[0]; element != nil && key == element.key {
        element.value = value
        return element
    }

    element = &Element{
        elementNode: elementNode{next: make([]*Element, list.randLevel())},  // 通过概率随机指定一个 level 给新插入的元素，调用了前面出现的函数 randLevel
        key:         key,
        value:       value,
    }
    list.length++

    for i := range element.next { // 插入数据
        element.next[i] = prevs[i].next[i]
        prevs[i].next[i] = element // 记录位置
    }

    return element
}

// Remove 获取key对应的值
func (list *SkipList) Remove(key float64) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var element *Element
    prevs := list.getPrevElementNodes(key)
    if element = prevs[0].next[0]; element != nil && key == element.key {
        for k, v := range element.next {
            prevs[k].next[k] = v // 删除
        }

        list.length--
        return element
    }

    return nil
}

Set

var element *Element
    prevs := list.getPrevElementNodes(key)
    if element = prevs[0].next[0]; element != nil && key == element.key {
        element.value = value
        return element
        }

getPrevElementNodeskey12prevs

6699

Setprevs12prevs[0].next[0]

if element = prevs[0].next[0]; element != nil && key == element.key

另外一块比较核心的代码就是：

    for i := range element.next { // 插入数据
        element.next[i] = prevs[i].next[i]
        prevs[i].next[i] = element // 记录位置
    }

SkipListprevNodesCacheRemove

RemoveSet

go 源代码

package skipList

import (
    "math"
    "math/rand"
    "sync"
    "time"
)

const (
    // DefaultMaxLevel 默认skip list最大深度
    DefaultMaxLevel int = 18
    // DefaultProbability 默认的概率
    DefaultProbability float64 = 1 / math.E
)

// elementNode 数组指针，指向元素
type elementNode struct {
    next []*Element
}

// Element 跳转表数据结构
type Element struct {
    elementNode
    key   float64
    value interface{} // 定义元素
}

// Key 获取key的值
func (e *Element) Key() float64 {
    return e.key
}

// Value 获取key的值
func (e *Element) Value() interface{} {
    return e.value
}


type SkipList struct {
    elementNode
    maxLevel       int            // 最大深度
    length         int            // 长度
    randSource     rand.Source    // 动态调节跳转表的长度
    probability    float64        // 概率
    probTable      []float64      // 存储位置，对应key
    mutex          sync.RWMutex   // 保证线程安全
    prevNodesCache []*elementNode // 缓存
}

// NewSkipList 新建跳转表
func NewSkipList() *SkipList {
    return NewWithMaxLevel(DefaultMaxLevel)
}

// ProbabilityTable 初始化 Probability Table
func ProbabilityTable(probability float64, maxLevel int) (table []float64) {
    for i := 1; i <= maxLevel; i++ {
        prob := math.Pow(probability, float64(i-1))
        table = append(table, prob)
    }
    return table
}

// NewWithMaxLevel 自定义maxLevel新建跳转表
func NewWithMaxLevel(maxLevel int) *SkipList {
    if maxLevel < 1 || maxLevel > DefaultMaxLevel {
        panic("invalid maxlevel")
    }

    return &SkipList{
        elementNode:    elementNode{next: make([]*Element, maxLevel)},
        prevNodesCache: make([]*elementNode, maxLevel),
        maxLevel:       maxLevel,
        randSource:     rand.New(rand.NewSource(time.Now().UnixNano())),
        probability:    DefaultProbability,
        probTable:      ProbabilityTable(DefaultProbability, maxLevel),
    }
}

// 随机计算最接近的
func (list *SkipList) randLevel() (level int) {
    r := float64(list.randSource.Int63()) / (1 << 63)
    level = 1
    for level < list.maxLevel && r < list.probTable[level] {
        level++ // 级别追加
    }

    return level
}

// SetProbability 设置新的概率,刷新概率表
func (list *SkipList) SetProbability(newProbability float64) {
    list.probability = newProbability
    list.probTable = ProbabilityTable(newProbability, list.maxLevel)
}

// Set 存储新的值
func (list *SkipList) Set(key float64, value interface{}) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var element *Element
    prevs := list.getPrevElementNodes(key)
    if element = prevs[0].next[0]; element != nil && key == element.key {
        element.value = value
        return element
    }

    element = &Element{
        elementNode: elementNode{next: make([]*Element, list.randLevel())},
        key:         key,
        value:       value,
    }
    list.length++

    for i := range element.next { // 插入数据
        element.next[i] = prevs[i].next[i]
        prevs[i].next[i] = element // 记录位置
    }

    return element
}

// Get 获取key对应的值
func (list *SkipList) Get(key float64) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var prev *elementNode = &list.elementNode // 保存前置结点
    var next *Element

    for i := list.maxLevel - 1; i >= 0; i-- {
        next = prev.next[i] // 循环跳到下一个
        for next != nil && key > next.key {
            prev = &next.elementNode
            next = next.next[i]
        }
    }

    if next != nil && next.key == key { // 找到
        return next
    }

    return nil // 没有找到
}

// Remove 获取key对应的值
func (list *SkipList) Remove(key float64) *Element {
    list.mutex.Lock()
    defer list.mutex.Unlock() // 线程安全

    var element *Element
    prevs := list.getPrevElementNodes(key)
    if element = prevs[0].next[0]; element != nil && key == element.key {
        for k, v := range element.next {
            prevs[k].next[k] = v // 删除
        }

        list.length--
        return element
    }

    return nil
}

func (list *SkipList) getPrevElementNodes(key float64) []*elementNode {
    var prev *elementNode = &list.elementNode // 保存前置结点
    var next *Element
    prevs := list.prevNodesCache // 缓冲集合
    for i := list.maxLevel - 1; i >= 0; i-- {
        next = prev.next[i] // 循环跳到下一个
        for next != nil && key > next.key {
            prev = &next.elementNode
            next = next.next[i]
        }
        prevs[i] = prev
    }
    return prevs
}