CloudGeek读源码系列-cache2go源码解析(二)
0、写在最前面
What:《cache2go的源码解析》会分为(一)(二)两讲,内容包括整个项目的所有功能代码和例子程序。
上一讲《》中已经分析了关键数据结构、cacheitem.go源码等部分,这一讲将继续分析剩下的所有代码及相关知识点。
一、代码逻辑
1、cachetable.go剩余部分
7.
上一讲说到CacheTable类型绑定的方法共计如上图,到SetLogger方法为止已经讲完了,这一讲从expirationCheck方法开始。
expirationCheck方法比较长,分析部分放在一起有点臃肿,所以我选择了源码加注释的方式来展示,每一行代码和相应的含义如下:
// Expiration check loop, triggered by a self-adjusting timer. // 【由计时器触发的到期检查】 func (table *CacheTable) expirationCheck() { table.Lock() //【计时器暂停】 if table.cleanupTimer != nil { table.cleanupTimer.Stop() } //【计时器的时间间隔】 if table.cleanupInterval > 0 { table.log("Expiration check triggered after", table.cleanupInterval, "for table", table.name) } else { table.log("Expiration check installed for table", table.name) } // To be more accurate with timers, we would need to update 'now' on every // loop iteration. Not sure it's really efficient though. //【当前时间】 now := time.Now() //【最小时间间隔,这里暂定义为0,下面代码会更新这个值】 smallestDuration := 0 * time.Second //【遍历一个table中的items】 for key, item := range table.items { // Cache values so we don't keep blocking the mutex. item.RLock() //【设置好的存活时间】 lifeSpan := item.lifeSpan //【最后一次访问的时间】 accessedOn := item.accessedOn item.RUnlock() //【存活时间为0的item不作处理,也就是一直存活】 if lifeSpan == 0 { continue } //【这个减法算出来的是这个item已经没有被访问的时间,如果比存活时间长,说明过期了,可以删了】 if now.Sub(accessedOn) >= lifeSpan { // Item has excessed its lifespan. //【删除操作】 table.deleteInternal(key) } else { // Find the item chronologically closest to its end-of-lifespan. //【按照时间顺序找到最接近过期时间的条目】 //【如果最后一次访问的时间到当前时间的间隔小于smallestDuration,则更新smallestDuration】 if smallestDuration == 0 || lifeSpan-now.Sub(accessedOn) < smallestDuration { smallestDuration = lifeSpan - now.Sub(accessedOn) } } } // Setup the interval for the next cleanup run. //【上面已经找到了最近接过期时间的时间间隔,这里将这个时间丢给了cleanupInterval】 table.cleanupInterval = smallestDuration //【如果是0就不科学了,除非所有条目都是0,那就不需要过期检测了】 if smallestDuration > 0 { //【计时器设置为smallestDuration,时间到则调用func这个函数】 table.cleanupTimer = time.AfterFunc(smallestDuration, func() { //这里并不是循环启动goroutine,启动一个新的goroutine后当前goroutine会退出,这里不会引起goroutine泄漏。 go table.expirationCheck() }) } table.Unlock() }
expirationCheck方法无非是做一个定期的数据过期检查操作,到目前为止这是项目中最复杂的一个方法,下面继续看剩下的部分。
8.
如上图所示,剩下的方法中划红线三个互相关联,我们放在一起看。
这次自上而下分析,明显Add和NotFoundAdd方法会调用addInternal方法,所以我们先看Add和NotFoundAdd方法。
先看Add()
注释部分说的很清楚,Add方法添加一个key/value对到cache,三个参数除了key、data、lifeSpan的含义我们在第一讲分析CacheItem类型的时候都已经介绍过。
NewCacheItem函数是cacheitem.go中定义的一个创建CacheItem类型实例的函数,返回值是*CacheItem类型。Add方法创建一个CacheItem类型实例后,将该实例的指针丢给了addInternal方法,然后返回了该指针。addInternal我们后面再看具体做了什么。
9.
大家注意到没有,这里的注释有一个单词写错了,they key应该是the key。
这个方法的参数和上面的Add方法是一样一样的,含义无需多说,方法体主要分2个部分:
开始的if判断是检查items中是否有这个key,存在则返回false;后面的代码自然就是不存在的时候执行的,创建一个CacheItem类型的实例,然后调用addInternal添加item,最后返回true;也就是说这个函数返回true是NotFound的情况。
ok,下面就可以看看addInternal这个方法干了啥了。
10.
这个方法无非是将CacheItem类型的实例添加到CacheTable中。方法开头的注释告诉我们调用这个方法前需要加锁,函数体前2行做了一个打日志和赋值操作,很好理解,然后将table.cleanupInterval和table.addedItem保存到局部变量,紧接着释放了锁。
后面的if部分调用了addedItem这个回调函数,也就是添加一个item时需要调用的函数。最后一个if判断稍微绕一点;
if的第一个条件:item.lifeSpan > 0,也就是当前item设置的存活时间是正数;然后&& (expDur == 0 || item.lifeSpan < expDur),expDur保存的是table.cleanupInterval,这个值为0也就是还没有设置检查时间间隔,或者item.lifeSpan < expDur也就是设置了,但是当前新增的item的lifeSpan要更小,这个时候就触发expirationCheck执行。这里可能有点绕,要注意lifeSpan是一个item的存活时间,而cleanupInterval是对于一个table来说触发检查还剩余的时间,如果前者更小,那么就说明需要提前出发check操作了。
11.
剩下的不多了,我们再看一组删除相关的方法
还是上面的套路,先看上层的调用者,当然就是Delete
接收一个key,调用deleteInternal(key)来完成删除操作,这里实在没有啥可讲的了,我们来看deleteInternal方法是怎么写的
12.
deleteInternal方法我也用详细注释的方式来解释吧~
func (table *CacheTable) deleteInternal(key interface{}) (*CacheItem, error) { r, ok := table.items[key] //【如果table中不存在key对应的item,则返回一个error】 //【ErrKeyNotFound在errors.go中定义,是errors.New("Key not found in cache")】 if !ok { return nil, ErrKeyNotFound } // Cache value so we don't keep blocking the mutex. //【将要删除的item缓存起来】 aboutToDeleteItem := table.aboutToDeleteItem table.Unlock() // Trigger callbacks before deleting an item from cache. //【删除操作执行前调用的回调函数,这个函数是CacheTable的属性,对应下面的是aboutToExpire是CacheItem的属性】 if aboutToDeleteItem != nil { aboutToDeleteItem(r) } r.RLock() defer r.RUnlock() //【这里对这条item加了一个读锁,然后执行了aboutToExpire回调函数,这个函数需要在item刚好要删除前执行】 if r.aboutToExpire != nil { r.aboutToExpire(key) } table.Lock() //【这里对表加了锁,上面已经对item加了读锁,然后这里执行delete函数删除了这个item】 //【delete函数是专门用来从map中删除特定key指定的元素的】 table.log("Deleting item with key", key, "created on", r.createdOn, "and hit", r.accessCount, "times from table", table.name) delete(table.items, key) return r, nil }
13.
万里长征最后几步咯~
最后5(Not打头这个咱说过了)个方法目测不难,咱一个一个来过,先看Exists
这里我是想说:来,咱略过吧~
算了,为了教程的完整性,还是简单说一下,读锁的相关代码不需要说了,剩下的只有一行:
_, ok := table.items[key]
这里如果key存在,ok为true,反之为false,就是这样,简单吧~
14.
Value()方法讲解,看注释吧~
// Value returns an item from the cache and marks it to be kept alive. You can // pass additional arguments to your DataLoader callback function. func (table *CacheTable) Value(key interface{}, args ...interface{}) (*CacheItem, error) { table.RLock() r, ok := table.items[key] //【loadData在load一个不存在的数据时被调用的回调函数】 loadData := table.loadData table.RUnlock() //【如果值存在,执行下面操作】 if ok { // Update access counter and timestamp. //【更新accessedOn为当前时间】 r.KeepAlive() return r, nil } //【这里当然就是值不存在的时候了】 // Item doesn't exist in cache. Try and fetch it with a data-loader. if loadData != nil { //【loadData这个回调函数是需要返回CacheItem类型的指针数据的】 item := loadData(key, args...) if item != nil { //【loadData返回了item的时候,万事大吉,执行Add】 table.Add(key, item.lifeSpan, item.data) return item, nil } //【item没有拿到,那就只能返回nil+错误信息了】 //【ErrKeyNotFoundOrLoadable是执行回调函数也没有拿到data的情况对应的错误类型】 return nil, ErrKeyNotFoundOrLoadable } //【这个return就有点无奈了,在loadData为nil的时候执行,也就是直接返回Key找不到】 return nil, ErrKeyNotFound }
15.
从注释可以看出来这个函数就是清空数据的作用,实现方式简单粗暴,让table的items属性指向一个新建的空map,cleanup操作对应的时间间隔设置为0,并且计时器停止。这里也可以得到cleanupInterval为0是什么场景,也就是说0不是代表清空操作死循环,间隔0秒就执行,而是表示不需要操作,缓存表还是空的。
16.
这个MostAccessed方法有点意思,涉及到sort.Sort的玩法,具体看下面注释:
// MostAccessed returns the most accessed items in this cache table //【访问频率高的count条item全部返回】 func (table *CacheTable) MostAccessed(count int64) []*CacheItem { table.RLock() defer table.RUnlock() //【这里的CacheItemPairList是[]CacheItemPair类型,是类型不是实例】 //【所以p是长度为len(table.items)的一个CacheItemPair类型的切片类型 p := make(CacheItemPairList, len(table.items)) i := 0 //【遍历items,将Key和AccessCount构造成CacheItemPair类型数据存入p切片】 for k, v := range table.items { p[i] = CacheItemPair{k, v.accessCount} i++ } //【这里可以直接使用Sort方法来排序是因为CacheItemPairList实现了sort.Interface接口,也就是Swap,Len,Less三个方法】 //【但是需要留意上面的Less方法在定义的时候把逻辑倒过来了,导致排序是从大到小的】 sort.Sort(p) var r []*CacheItem c := int64(0) for _, v := range p { //【控制返回值数目】 if c >= count { break } item, ok := table.items[v.Key] if ok { //【因为数据是按照访问频率从高到底排序的,所以可以从第一条数据开始加】 r = append(r, item) } c++ } return r }
17.
最后一个方法了,哇咔咔,好长啊~~~
这个函数也没有太多可以讲的,为了方便而整的内部日志函数,日志相关操作我会在《CloudGeek讲golang系列》中单独拿出一讲来详细介绍。
2、cache.go
前面的代码看完之后现在看cache.go就太简单了,上代码吧~
var ( cache = make(map[string]*CacheTable) mutex sync.RWMutex ) // Cache returns the existing cache table with given name or creates a new one // if the table does not exist yet. //【表存在则返回表,表不存在的时候创建一个不包含items的空表,然后返回之】 func Cache(table string) *CacheTable { mutex.RLock() //【注意cache的类型,这是一个用于存CacheTable的map】 t, ok := cache[table] mutex.RUnlock() if !ok { mutex.Lock() //【表不存在的时候需要创建一个空表,这时候做了一个二次检查,为的是并发安全】 t, ok = cache[table] // Double check whether the table exists or not. if !ok { t = &CacheTable{ name: table, items: make(map[interface{}]*CacheItem), } cache[table] = t } mutex.Unlock() } return t }
二、example
如上图,项目中还有一个examples目录,大家肯定已经猜到了里面的内容了,没错,就是上面介绍的一堆代码组成的缓存库,怎么玩?
下面我们一个一个来看这3个示例吧~
1、callbacks.go
func main() { //【创建一个名为myCache的缓存表】 cache := cache2go.Cache("myCache") // This callback will be triggered every time a new item // gets added to the cache. //【每次有新item被加入到这个缓存表的时候会被触发的回调函数】 //【这个函数只做了一个输出的动作】 cache.SetAddedItemCallback(func(entry *cache2go.CacheItem) { fmt.Println("Added:", entry.Key(), entry.Data(), entry.CreatedOn()) }) // This callback will be triggered every time an item // is about to be removed from the cache. //【当一个item被删除时被触发执行的回调函数,同样只有一个打印功能】 cache.SetAboutToDeleteItemCallback(func(entry *cache2go.CacheItem) { fmt.Println("Deleting:", entry.Key(), entry.Data(), entry.CreatedOn()) }) // Caching a new item will execute the AddedItem callback. //【缓存中添加一条记录】 cache.Add("someKey", 0, "This is a test!") // Let's retrieve the item from the cache //【读取刚才存入的数据】 res, err := cache.Value("someKey") if err == nil { fmt.Println("Found value in cache:", res.Data()) } else { fmt.Println("Error retrieving value from cache:", err) } // Deleting the item will execute the AboutToDeleteItem callback. //【删除someKey对应的记录】 cache.Delete("someKey") // Caching a new item that expires in 3 seconds //【添加设置了3s存活时间的记录】 res = cache.Add("anotherKey", 3*time.Second, "This is another test") // This callback will be triggered when the item is about to expire //【一旦触发了删除操作就会调用到下面这个回调函数,在这里也就是3s到期时被执行 res.SetAboutToExpireCallback(func(key interface{}) { fmt.Println("About to expire:", key.(string)) }) //【为了等上面的3s时间到】 time.Sleep(5 * time.Second) }
2、dataloader.go
func main() { cache := cache2go.Cache("myCache") // The data loader gets called automatically whenever something // tries to retrieve a non-existing key from the cache. //【当从cache中访问一个不存在的key时会触发这个回调函数】 cache.SetDataLoader(func(key interface{}, args ...interface{}) *cache2go.CacheItem { // Apply some clever loading logic here, e.g. read values for // this key from database, network or file. //【这里可以做一些机智的处理,比如说从数据库,网络或者文件中读取数据,这当然也是缓存的意义】 //key.(string)是类型断言,将interface{}类型的数据转回到string类型 val := "This is a test with key " + key.(string) // This helper method creates the cached item for us. Yay! //【很棒,这样就构造了一个value,然后构造出item】 item := cache2go.NewCacheItem(key, 0, val) return item }) // Let's retrieve a few auto-generated items from the cache. //【试着检索一些自动生成的items】 for i := 0; i < 10; i++ { //【将i转换为字符串,拼接成someKey_1的形式】 res, err := cache.Value("someKey_" + strconv.Itoa(i)) if err == nil { fmt.Println("Found value in cache:", res.Data()) } else { fmt.Println("Error retrieving value from cache:", err) } } }
3、mycachedapp.go
// Keys & values in cache2go can be of arbitrary types, e.g. a struct. //【这个例子中要存储的数据是如下结构体类型】 type myStruct struct { text string moreData []byte } func main() { // Accessing a new cache table for the first time will create it. //【创建缓存表myCache】 cache := cache2go.Cache("myCache") // We will put a new item in the cache. It will expire after // not being accessed via Value(key) for more than 5 seconds. //【构造一个数据】 val := myStruct{"This is a test!", []byte{}} //【存入数据,设置存活时间为5s】 cache.Add("someKey", 5*time.Second, &val) // Let's retrieve the item from the cache. //【试着读取】 res, err := cache.Value("someKey") if err == nil { fmt.Println("Found value in cache:", res.Data().(*myStruct).text) } else { fmt.Println("Error retrieving value from cache:", err) } // Wait for the item to expire in cache. //【等待6s之后,明显是该过期了】 time.Sleep(6 * time.Second) res, err = cache.Value("someKey") if err != nil { fmt.Println("Item is not cached (anymore).") } // Add another item that never expires. //【再存入一个永不过期的数据】 cache.Add("someKey", 0, &val) // cache2go supports a few handy callbacks and loading mechanisms. //【设置回调函数,删除数据的时候被调用】 cache.SetAboutToDeleteItemCallback(func(e *cache2go.CacheItem) { fmt.Println("Deleting:", e.Key(), e.Data().(*myStruct).text, e.CreatedOn()) }) // Remove the item from the cache. //【手动执行删除操作,当然,上面设置的函数将被调用】 cache.Delete("someKey") // And wipe the entire cache table. //【抹去整个cache中的数据】 cache.Flush() }
终于全部讲完了~
如上,整个项目除了测试代码外,实现逻辑部分全部分析完了,希望对你有帮助!