LevelDB的cache---golang实现

2022-04-18 00:00:00 索引 返回 获取 链表 双向

前言

对LevelDB比较了解应该知道其中一块cache实现:LRU Cache是采用双向链表 + HashTable,其中HashTable是为了解决双向链表的查找性能缺陷才引入的。接下来看一下采用golang如何实现LRU Cache。

实现

1、定义cache接口【thread-safe的cache】
import (        "io")// 参考:https://sourcegraph.com/github.com/google/leveldb/-/blob/include/leveldb/cache.htype Cache interface{  // 在multiple clients下共享相同的cache来划分key空间
  // 当启动是会被分配一个新id 
  NewId()  uint64 
  
  // 插入: 从 key-value到cache的映射
  // 当insert的entry不在需要时对应的key-value传给deleter处理
  Insert(key string, value interface{}, size int, deleter func(key string, value interface{})) (handle io.Closer)  // 查找: 不存在时返回 nil,nil, false
  Lookup(key string) (value interface{}, handle io.Closer, ok bool)  // 删除: 注意必须所有关联的handle释放才能erase key
  Erase(key string)  // 关闭: 所有的handle必须release 才能destroy所有存在的entries
  Close()  error
}

2、LRUCache

import (  "container/list"
  "fmt"
  "io"
  "runtime"
  "sync"
  "sync/atomic"
  "time")

var _Cache = (*LRUCache)(nil)// LRU cache实现type   LRUCache struct{
    * _LRUCache
}

type _LRUCache struct{
   mu  sync.Mutex   //  采用双向链表  和 HashTable
   //  双向循环链表,有大小限制,保证数据的新旧,当缓存不够时,保证先清除旧的数据
   list     *list.List   // 二级指针数组,链表没有大小限制,动态扩展大小,保证数据快速查找,hash定位一级指针,得到存放在一级指针上的二级指针链表,遍历查找数据
   table  map[string]*list.Element   // 当前size
   // cache容量
   size         int64
   capacity  int64
   last_id     uint64
}// handle to an entry: // 哈希表中的元素type LRUHandle struct{
    c                      *LRUCache
    key                   string
    value                interface{}
    size                   int64
    deleter              func(key string, vlaue interface{})
    time_created    time.Time
    time_accessed atomic.Value
    refs                    uint32
}//  获取keyfunc   (h *LRUHandle)  Key()  string{  return h.key
}// 获取valuefunc (h *LRUHandle) Value() interface{} {    return h.value
}// 获取sizefunc (h *LRUHandle) Size() int {    return int(h.size)
}// 获取entry创建时间func (h *LRUHandle) TimeCreated() time.Time {    return h.time_created
}// 获取entry访问时间func (h *LRUHandle) TimeAccessed() time.Time {    return h.time_accessed.Load().(time.Time)
}// func (h *LRUHandle) Retain() (handle *LRUHandle) {
    h.c.mu.Lock()
    defer h.c.mu.Unlock()
    h.c.addref(h)    return h
}

func (h *LRUHandle) Close() error {
    h.c.mu.Lock()
    defer h.c.mu.Unlock()
    h.c.unref(h)    return nil
}

=================华丽分割线=================// 根据指定capacity新建一个cachefunc NewLRUCache(capacity int64) *LRUCache {
    assert(capacity > )
    p := &_LRUCache{        list:     list.New(),
        table:    make(map[string]*list.Element),
        capacity: capacity,
    }
    runtime.SetFinalizer(p, (*_LRUCache).Close)    return &LRUCache{p}
}// 清除所有已存在的entry通过deleter函数(用户自定义)// 前提:所有的handle必须releasefunc (p *LRUCache) Close() error {
    runtime.SetFinalizer(p._LRUCache, nil)
    p._LRUCache.Close()    return nil
}// 获取cache中key对应的内容 func (p *LRUCache) Get(key string) (value interface{}, ok bool) {    if v, h, ok := p.Lookup(key); ok {
        h.Close()        return v, ok
    }    return}// 类似get,差异在于cache中不存在时,可交由用户自行处理返回结果同时将返回的结果存放到cache中func (p *LRUCache) GetFrom(key string, getter func(key string) (v interface{}, size int, err error)) (value interface{}, err error) {    if v, h, ok := p.Lookup(key); ok {
        h.Close()        return v, nil
    }    if getter == nil {        return nil, fmt.Errorf("cache: %q not found!", key)
    }
    value, size, err := getter(key) // 由使用方自行处理 比如从其他存储介质获取
    if err != nil {        return
    }
    assert(size > )
    p.Set(key, value, size)   // 将内容添加到cache
    return}// 获取value并指定默认值 当cache中不存在时返回个default值作为结果func (p *LRUCache) Value(key string, defaultValue ...interface{}) interface{} {    if v, h, ok := p.Lookup(key); ok {
        h.Close()        return v
    }    if len(defaultValue) >  {        return defaultValue[]
    } else {        return nil
    }
}// 添加key-value并指定size同时也设定deleter用于当key-value不用时执行deletefunc (p *LRUCache) Set(key string, value interface{}, size int, deleter ...func(key string, value interface{})) {    if len(deleter) >  {  // 给定deleter函数
        h := p.Insert(key, value, size, deleter[])
        h.Close()
    } else {
        h := p.Insert(key, value, size, nil)
        h.Close()
    }
}// 能够用于多clients共享相同的cache来分割key空间// 一般在启动时会分配一个新的id给client,可将id作为key的前缀func (p *LRUCache) NewId() uint64 {
    p.mu.Lock()
    defer p.mu.Unlock()

    p.last_id++    return p.last_id
}// 建立key-value到cache的映射并分配指定size//  返回操作该映射的handle 当映射不再被需要时handle.Close会被调用// 另外当insert的entry不再需要会触发deleterfunc (p *LRUCache) Insert(key string, value interface{}, size int, deleter func(key string, value interface{})) (handle io.Closer) {
    handle = p.Insert_(key, value, size, deleter)    return}// 类似insert 不过返回值=LRUHandlefunc (p *LRUCache) Insert_(key string, value interface{}, size int, deleter func(key string, value interface{})) (handle *LRUHandle) {
    p.mu.Lock()
    defer p.mu.Unlock()

    assert(key != "" && size > )       // 当二级指针数组中已存在对应的key
       // 需要先执行remove在table和list
      // 当缓存不够时  保证cache数据新
    if element := p.table[key]; element != nil {
        p.list.Remove(element)  // 删除双向链表中element
        delete(p.table, key) // 删除二级索引表

        h := element.Value.(*LRUHandle) 
        p.unref(h)                     // 解除引用
    } 
        
        // 新建LRUHandle
    h := &LRUHandle{
        c:            p,
        key:          key,
        value:        value,
        size:         int64(size),
        deleter:      deleter,
        time_created: time.Now(),
        refs:         2, // One from LRUCache, one for the returned handle
    }
    h.time_accessed.Store(time.Now())        // 放置表头 新的数据
    element := p.list.PushFront(h)
    p.table[key] = element
    p.size += h.size
    p.checkCapacity()  // 检查是否超过capacity
    return h
}// 查找// 若是cache中不存在对应key的内容返回nil,nil,false// 否则返回handle(key-value与cache的映射)// 当mapping不存需要时 需要调用handle.Close() func (p *LRUCache) Lookup(key string) (value interface{}, handle io.Closer, ok bool) {    // warning: (*LRUHandle)(nil) != (io.Closer)(nil)
    if v, h, ok := p.Lookup_(key); ok {        return v, h, ok
    }    return}// 类似Lookup 但返回值 *LRUHandle.func (p *LRUCache) Lookup_(key string) (value interface{}, handle *LRUHandle, ok bool) {
    p.mu.Lock()
    defer p.mu.Unlock()        // 通过二级索引数组 获取key的内容
    element := p.table[key]    if element == nil {        return nil, nil, false
    }        // 通过二级索引table查找到element 再将双向链表list中的element移到表头
    p.list.MoveToFront(element)       // 获取到对应的元素
    h := element.Value.(*LRUHandle)
    h.time_accessed.Store(time.Now())       // 记录当前handle的引用数
    p.addref(h)    return h.Value(), h, true}// 获取key对应的内容 并删除cache中双向链表和二级索引的内容func (p *LRUCache) Take(key string) (handle io.Closer, ok bool) {
    p.mu.Lock()
    defer p.mu.Unlock()

    element := p.table[key]    if element == nil {        return nil, false
    }        // 删除双向链表对应的element
    p.list.Remove(element)       // 删除二级索引中的记录
    delete(p.table, key)

    h := element.Value.(*LRUHandle)    return h, true}// 删除key : 注意需要release所有关联的handlefunc (p *LRUCache) Erase(key string) {
    p.mu.Lock()
    defer p.mu.Unlock()

    element := p.table[key]    if element == nil {        return
    }        // 删除list中的element和table中记录
    p.list.Remove(element)    delete(p.table, key)
       
    h := element.Value.(*LRUHandle)
    p.unref(h)       // 释放与handle的关联
    return}// 设置cache的capacity// 当capacity过小且当前cache size远超过capacity时,cache将进行收缩func (p *LRUCache) SetCapacity(capacity int64) {
    p.mu.Lock()
    defer p.mu.Unlock()

    assert(capacity > )
    p.capacity = capacity
    p.checkCapacity()
}// 返回cache的少量统计信息:// 当前entry的数量、cache size、cache capacity、久访问entry的时间func (p *LRUCache) Stats() (length, size, capacity int64, oldest time.Time) {
    p.mu.Lock()
    defer p.mu.Unlock()    if lastElem := p.list.Back(); lastElem != nil {
        oldest = lastElem.Value.(*LRUHandle).time_accessed.Load().(time.Time)
    }    return int64(p.list.Len()), p.size, p.capacity, oldest
}// 统计信息以json串显示func (p *LRUCache) StatsJSON() string {    if p == nil {        return "{}"
    }
    l, s, c, o := p.Stats()    return fmt.Sprintf(`{    "Length": %v,    "Size": %v,    "Capacity": %v,    "OldestAccess": "%v"}`, l, s, c, o)
}// cache中element的数量:func (p *LRUCache) Length() int64 {
    p.mu.Lock()
    defer p.mu.Unlock()    return int64(p.list.Len())
}// 返回cache的sizefunc (p *LRUCache) Size() int64 {
    p.mu.Lock()
    defer p.mu.Unlock()    return p.size
}// 返回cache的capacityfunc (p *LRUCache) Capacity() int64 {
    p.mu.Lock()
    defer p.mu.Unlock()    return p.capacity
}// 返回cache新访问的element的time// 若是cache为空 则返回IsZero()的时间func (p *LRUCache) Newest() (newest time.Time) {
    p.mu.Lock()
    defer p.mu.Unlock()    if frontElem := p.list.Front(); frontElem != nil {
        newest = frontElem.Value.(*LRUHandle).time_accessed.Load().(time.Time)
    }    return}// 返回cache久访问的element的time// 若是cache为空 则返回IsZero()的时间func (p *LRUCache) Oldest() (oldest time.Time) {
    p.mu.Lock()
    defer p.mu.Unlock()    if lastElem := p.list.Back(); lastElem != nil {
        oldest = lastElem.Value.(*LRUHandle).time_accessed.Load().(time.Time)
    }    return}// 返回cache中新使用key集合func (p *LRUCache) Keys() []string {
    p.mu.Lock()
    defer p.mu.Unlock()

    keys := make([]string, , p.list.Len())    for e := p.list.Front(); e != nil; e = e.Next() {
        keys = append(keys, e.Value.(*LRUHandle).key)
    }    return keys
}// LRUHandle引用计数(增加)func (p *_LRUCache) addref(h *LRUHandle) {
    h.refs++
}// LRUHandle引用计数(减少)func (p *_LRUCache) unref(h *LRUHandle) {
    assert(h.refs > )
    h.refs--    if h.refs <=  {
        p.size -= h.size        if h.deleter != nil {
            h.deleter(h.key, h.value)
        }
    }
}// 检查capacity:压缩相对时间比较久的entry(剔除)func (p *LRUCache) checkCapacity() {        // 当cache的size大于capacity同时二级索引数组不止1个 进行shrank
       // 从双向链表的尾部开始remove对应的element
       // 同时二级索引数组对应的内容也需要delete 并减少对应的引用计数 
      // 直至size <= capacity 二级索引数组个数 = 1
    for p.size > p.capacity && len(p.table) > 1 {
        delElem := p.list.Back()
        h := delElem.Value.(*LRUHandle)
        p.list.Remove(delElem)        delete(p.table, h.key)
        p.unref(h)
    }
}// 清空func (p *LRUCache) Clear() {
    p.mu.Lock()
    defer p.mu.Unlock()        // release 二级索引数组关联的handle
    for _, element := range p.table {
        h := element.Value.(*LRUHandle)
        p.unref(h)
    }        // cache的双向链表和二级索引数组置空
    p.list = list.New()
    p.table = make(map[string]*list.Element)
    p.size = 
    return}//   cache的closefunc (p *_LRUCache) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()    for _, element := range p.table {
        h := element.Value.(*LRUHandle)
        assert(h.refs == 1, "h.refs = ", h.refs)
        p.unref(h)
    }

    p.list = nil
    p.table = nil
    p.size = }
至此lru cache实现完成

3、应用

package mainimport (    "fmt"
    "cache")func main() {        // 新建cache
    c := cache.NewLRUCache(100)
    defer c.Close()        // set entry到缓存
    c.Set("key1", "value1", 1)
    value1 := c.Value("key1").(string)
    fmt.Println("key1:", value1)

    c.Set("key2", "value2", 1)
    value2 := c.Value("key2", "null").(string) // 指定默认值
    fmt.Println("key2:", value2)

    value3 := c.Value("key3", "null").(string)
    fmt.Println("key3:", value3)

    value4 := c.Value("key4") // value4 is nil
    fmt.Println("key4:", value4)

    fmt.Println("Done")
}

Output:

key1: value1
key2: value2
key3: nullkey4: <nil>Done

Non GC Object

package mainimport (    "fmt"
    "cache")func main() {
    c := cache.NewLRUCache(10)
    defer c.Close()

    id0 := c.NewId()
    id1 := c.NewId()
    id2 := c.NewId()
    fmt.Println("id0:", id0)
    fmt.Println("id1:", id1)
    fmt.Println("id2:", id2)    // add new
    v1 := "data:123"
    h1 := c.Insert("123", "data:123", len("data:123"), func(key string, value interface{}) {
        fmt.Printf("deleter(%q:%q)\n", key, value)
    })    // fetch ok
    v2, h2, ok := c.Lookup("123")    assert(ok)    assert(h2 != nil)    // remove
    c.Erase("123")    // fetch failed
    _, h3, ok := c.Lookup("123")    assert(!ok)    assert(h3 == nil)    // h1&h2 still valid!
    fmt.Printf("user1(%s)\n", v1)
    fmt.Printf("user2(%s)\n", v2.(string))    // release h1
        // 由于 h2  handle对应值 故而deleter不能被触发
    h1.Close()    // invoke deleter
    fmt.Println("invoke deleter(123) begin")
    h2.Close()
    fmt.Println("invoke deleter(123) end")    // 添加
    h4 := c.Insert("abc", "data:abc", len("data:abc"), func(key string, value interface{}) {
        fmt.Printf("deleter(%q:%q)\n", key, value)
    })    // release h4
    // 由于cache在handle对应值 故而deleter不能被触发
    h4.Close()    // cache length
    length := c.Length()    assert(length == 1)    // cache size
    size := c.Size()    assert(size == 8, "size:", size)    // add h5
        //  超过容量10 会触发h4的deleter
    fmt.Println("invoke deleter(h4) begin")
    h5 := c.Insert("456", "data:456", len("data:456"), func(key string, value interface{}) {
        fmt.Printf("deleter(%q:%q)\n", key, value)
    })
    fmt.Println("invoke deleter(h4) end")    // 必须释放所有的handle
    h5.Close()    // stats
    fmt.Println("StatsJSON:", c.StatsJSON())    // done
    fmt.Println("Done")
}func assert(v bool, a ...interface{}) {    if !v {
        panic(fmt.Sprint(a...))
    }
}

输出:

id0: 1id1: 2id2: 3user1(data:123)
user2(data:123)
invoke deleter(123) begindeleter("123":"data:123")
invoke deleter(123) endinvoke deleter(h4) begindeleter("abc":"data:abc")
invoke deleter(h4) endStatsJSON: {        "Length": 1,        "Size": 8,        "Capacity": 10,        "OldestAccess": "2015-08-21 18:00:24.0119469 +0800 CST"}
Done
deleter("456":"data:456")

来自:https://mp.weixin.qq.com/s/aLiwPuhF4onHon3VfeAVMw

相关文章