前言
对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.h
type 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
}
// 获取key
func (h *LRUHandle) Key() string{
return h.key
}
// 获取value
func (h *LRUHandle) Value() interface{} {
return h.value
}
// 获取size
func (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新建一个cache
func NewLRUCache(capacity int64) *LRUCache {
assert(capacity > 0)
p := &_LRUCache{
list: list.New(),
table: make(map[string]*list.Element),
capacity: capacity,
}
runtime.SetFinalizer(p, (*_LRUCache).Close)
return &LRUCache{p}
}
// 清除所有已存在的entry通过deleter函数(用户自定义)
// 前提：所有的handle必须release
func (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 > 0)
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) > 0 {
return defaultValue[0]
} else {
return nil
}
}
// 添加key-value并指定size同时也设定deleter用于当key-value不用时执行delete
func (p *LRUCache) Set(key string, value interface{}, size int, deleter ...func(key string, value interface{})) {
if len(deleter) > 0 { // 给定deleter函数
h := p.Insert(key, value, size, deleter[0])
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不再需要会触发deleter
func (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 不过返回值=LRUHandle
func (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 > 0)
// 当二级指针数组中已存在对应的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所有关联的handle
func (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 > 0)
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的size
func (p *LRUCache) Size() int64 {
p.mu.Lock()
defer p.mu.Unlock()
return p.size
}
// 返回cache的capacity
func (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, 0, 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 > 0)
h.refs--
if h.refs <= 0 {
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 = 0
return
}
// cache的close
func (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 = 0
}
至此lru cache实现完成
3、应用
package main
import (
"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: null
key4: <nil>
Done
Non GC Object
package main
import (
"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: 1
id1: 2
id2: 3
user1(data:123)
user2(data:123)
invoke deleter(123) begin
deleter("123":"data:123")
invoke deleter(123) end
invoke deleter(h4) begin
deleter("abc":"data:abc")
invoke deleter(h4) end
StatsJSON: {
"Length": 1,
"Size": 8,
"Capacity": 10,
"OldestAccess": "2015-08-21 18:00:24.0119469 +0800 CST"
}
Done
deleter("456":"data:456")
有疑问加站长微信联系（非本文作者）
