Mango Cache缓存管理库TinyLFU源码解析_Golang_

type localCache struct { cache cache // 真正存放数据的地方 expireAfterAccess time.Duration //用户自定义的过期以及刷新时间 expireAfterWrite time.Duration refreshAfterWrite time.Duration policyName string //缓存管理策略 onInsertion Func //钩子函数 onRemoval Func loader LoaderFunc //自定义加载和重载函数 reloader Reloader stats StatsCounter //状态管理器 cap int //缓存容量 // 访问时的缓存管理策略 accessQueue policy // 写入时的缓存管理策略,只有在expireAfterWrite或refreshAfterWrite被设置以后才启用 writeQueue policy // 用于processEntries的事件队列 events chan entryEvent // 记录距离上一次写期间发生的读次数,用于判断是否进行清扫 readCount int32 // 用于关闭由localCache创建的协程 closing int32 closeWG sync.WaitGroup } 

type cache struct { size int64 // 缓存大小 segs [segmentCount]sync.Map // 分片sync.map, map[key]*entry } 

type entry struct { hash uint64 //key的hash值 accessTime int64 writeTime int64 invalidated int32 //是否有效 loading int32 //是否处于载入中状态 key Key value atomic.Value // 存储的值 // 下面的属性只被缓存管理策略操作,所以不需要原子性的访问 accessList *list.Element //此entry目前所在的访问链表中的位置 writeList *list.Element //此entry目前所在的写入链表中的位置 listID uint8 //此entry目前所在的缓存段(SLRU) } 

func NewLoadingCache(loader LoaderFunc, options ...Option) LoadingCache { c := newLocalCache() c.loader = loader for _, opt := range options { opt(c) } c.init() return c } 

func (c *localCache) init() { c.accessQueue = newPolicy(c.policyName) c.accessQueue.init(&c.cache, c.cap) if c.expireAfterWrite > 0 || c.refreshAfterWrite > 0 { c.writeQueue = &recencyQueue{} } else { c.writeQueue = discardingQueue{} } c.writeQueue.init(&c.cache, c.cap) c.events = make(chan entryEvent, chanBufSize) c.closeWG.Add(1) go c.processEntries() } 

func (c *localCache) Get(k Key) (Value, error) { en := c.cache.get(k, sum(k)) //计算key的hash并查询该key if en == nil { c.stats.RecordMisses(1) return c.load(k) } // 检查entry是否需要更新 now := currentTime() if c.isExpired(en, now) { if c.loader == nil { //如果没有设置加载器则直接删除该entry c.sendEvent(eventDelete, en) } else { //对于loadingCache,我们不删除这个entry //而是把它暂时留在缓存中,所以用户依旧可以读取到旧的缓存值 c.setEntryAccessTime(en, now) c.refreshAsync(en) } c.stats.RecordMisses(1) } else { c.setEntryAccessTime(en, now) c.sendEvent(eventAccess, en) c.stats.RecordHits(1) } return en.getValue(), nil } 

func (c *localCache) refreshAsync(en *entry) bool { if en.setLoading(true) { // 如果此entry没有在加载 if c.reloader == nil { go c.refresh(en) } else { c.reload(en) } return true } return false } 

func (c *localCache) Put(k Key, v Value) { h := sum(k) en := c.cache.get(k, h) now := currentTime() if en == nil { en = newEntry(k, v, h) c.setEntryWriteTime(en, now) c.setEntryAccessTime(en, now) // 直接将新value添加进缓存(在缓存容量足够的时候) if c.cap == 0 || c.cache.len() < c.cap { cen := c.cache.getOrSet(en) if cen != nil { cen.setValue(v) c.setEntryWriteTime(cen, now) en = cen } } } else { // 更新entry en.setValue(v) c.setEntryWriteTime(en, now) } c.sendEvent(eventWrite, en) } //当entry在缓存中存在,则返回该entry,否则存储该entry func (c *cache) getOrSet(v *entry) *entry { seg := c.segment(v.hash) en, ok := seg.LoadOrStore(v.key, v) if ok { return en.(*entry) } atomic.AddInt64(&c.size, 1) return nil } 

type event uint8 const ( eventWrite event = iota eventAccess eventDelete eventClose ) type entryEvent struct { entry *entry event event } func (c *localCache) processEntries() { defer c.closeWG.Done() for e := range c.events { switch e.event { case eventWrite: //写入事务 c.write(e.entry) c.postWriteCleanup() //清理操作 case eventAccess: //访问事务 c.access(e.entry) c.postReadCleanup() //清理操作 case eventDelete: if e.entry == nil { //InvalidateAll函数中使用 c.removeAll() } else { c.remove(e.entry) //移除单个entry } c.postReadCleanup() //清理操作 case eventClose: if c.reloader != nil { // 停止所有refresh工作 c.reloader.Close() } c.removeAll() return } } } 

func (c *localCache) write(en *entry) { ren := c.accessQueue.write(en) c.writeQueue.write(en) if c.onInsertion != nil { c.onInsertion(en.key, en.getValue()) } if ren != nil { //有entry被驱逐出了缓存 c.writeQueue.remove(ren) c.stats.RecordEviction() if c.onRemoval != nil { c.onRemoval(ren.key, ren.getValue()) } } } 

func (c *localCache) postReadCleanup() { if atomic.AddInt32(&c.readCount, 1) > drainThreshold { atomic.StoreInt32(&c.readCount, 0) c.expireEntries() } } // 在添加完entry以后再执行 func (c *localCache) postWriteCleanup() { atomic.StoreInt32(&c.readCount, 0) c.expireEntries() } //清理工作函数 func (c *localCache) expireEntries() { remain := drainMax now := currentTime() if c.expireAfterAccess > 0 { expiry := now.Add(-c.expireAfterAccess).UnixNano() c.accessQueue.iterate(func(en *entry) bool { if remain == 0 || en.getAccessTime() >= expiry { // 找到了第一个没有过期的entry或者清理entry数足够了 return false } c.remove(en) c.stats.RecordEviction() remain-- return remain > 0 }) } if remain > 0 && c.expireAfterWrite > 0 { ... } if remain > 0 && c.loader != ni
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：golang服务报错: write: broken pipe的解决方案_Golang_
                                                                
下一篇：LRU LFU TinyLFU缓存算法实例详解_Golang_