diff --git a/g/container/gmap/string_interface_map.go b/g/container/gmap/string_interface_map.go index 144152b4d..4fb7e17a5 100644 --- a/g/container/gmap/string_interface_map.go +++ b/g/container/gmap/string_interface_map.go @@ -83,13 +83,13 @@ func (this *StringInterfaceMap) GetWithDefault(key string, value interface{}) in func (this *StringInterfaceMap) GetOrSetFunc(key string, f func() interface{}) interface{} { if v := this.Get(key); v == nil { + value := f() this.mu.Lock() defer this.mu.Unlock() // 写锁二次检索确认 if v, ok := this.m[key]; !ok { - v = f() - this.m[key] = v - return v + this.m[key] = value + return value } else { return v } diff --git a/g/container/gqueue/gqueue.go b/g/container/gqueue/gqueue.go index fddac93c7..f728344bb 100644 --- a/g/container/gqueue/gqueue.go +++ b/g/container/gqueue/gqueue.go @@ -61,6 +61,7 @@ func (q *Queue) startAsyncLoop() { case <- q.closeChan: return case <- q.events: + // 循环读取链表，直到为空才跳出 for { if v := q.list.PopFront(); v != nil { q.queue <- v @@ -78,7 +79,9 @@ func (q *Queue) Push(v interface{}) { q.queue <- v } else { q.list.PushBack(v) - q.events <- struct{}{} + if len(q.events) == 0 { + q.events <- struct{}{} + } } } diff --git a/g/os/gcache/gcache_mem_cache.go b/g/os/gcache/gcache_mem_cache.go index 7b5ed299f..79af2ccb6 100644 --- a/g/os/gcache/gcache_mem_cache.go +++ b/g/os/gcache/gcache_mem_cache.go @@ -26,7 +26,8 @@ type memCache struct { data map[interface{}]memCacheItem // 缓存数据(所有的缓存数据存放哈希表) ekmap map[interface{}]int64 // 键名对应的分组过期时间(用于相同键名过期时间快速更新)，键值为10秒级时间戳 eksets map[int64]*gset.Set // 分组过期时间对应的键名列表(用于自动过期快速删除)，键值为10秒级时间戳 - eventQueue *glist.List // 异步处理队列 + eventList *glist.List // 异步处理队列 + lruGetList *glist.List // 获取方法的LRU列表 stopChan chan struct{} // 关闭时间通知 } @@ -38,8 +39,8 @@ type memCacheItem struct { // 异步队列数据项 type memCacheEvent struct { - k interface{} // 键名 - e int64 // 过期时间 + k interface{} // 键名 + e int64 // 过期时间 } // 创建底层的缓存对象 @@ -50,7 +51,8 @@ func newMemCache(lruCap...int) *memCache { ekmap : make(map[interface{}]int64), eksets : make(map[int64]*gset.Set), stopChan : make(chan struct{}), - eventQueue : glist.New(), + eventList : glist.New(), + lruGetList : glist.New(), } if len(lruCap) > 0 { c.cap = lruCap[0] @@ -98,11 +100,7 @@ func (c *memCache) Set(key interface{}, value interface{}, expire int) { c.dmu.Lock() c.data[key] = memCacheItem{v : value, e : expireTimestamp} c.dmu.Unlock() - c.eventQueue.PushBack(memCacheEvent{k : key, e : expireTimestamp}) - // LRU(Least Recently Used)操作记录 - if c.cap > 0 { - c.lru.Push(key) - } + c.eventList.PushBack(memCacheEvent{k : key, e : expireTimestamp}) } // 设置kv缓存键值对，内部会对键名的存在性使用写锁进行二次检索确认，如果存在则不再写入；返回键名对应的键值。 @@ -116,10 +114,7 @@ func (c *memCache) doSetWithLockCheck(key interface{}, value interface{}, expire } c.data[key] = memCacheItem{v : value, e : expireTimestamp} c.dmu.Unlock() - c.eventQueue.PushBack(memCacheEvent{k : key, e : expireTimestamp}) - if c.cap > 0 { - c.lru.Push(key) - } + c.eventList.PushBack(memCacheEvent{k : key, e : expireTimestamp}) return value } @@ -150,10 +145,7 @@ func (c *memCache) BatchSet(data map[interface{}]interface{}, expire int) { c.dmu.Lock() c.data[k] = memCacheItem{v: v, e: expireTimestamp} c.dmu.Unlock() - c.eventQueue.PushBack(memCacheEvent{k: k, e: expireTimestamp}) - if c.cap > 0 { - c.lru.Push(k) - } + c.eventList.PushBack(memCacheEvent{k: k, e: expireTimestamp}) } } @@ -165,7 +157,7 @@ func (c *memCache) Get(key interface{}) interface{} { if ok && !item.IsExpired() { // LRU(Least Recently Used)操作记录 if c.cap > 0 { - c.lru.Push(key) + c.lruGetList.PushBack(key) } return item.v } @@ -267,34 +259,56 @@ func (c *memCache) Close() { // 数据自动同步循环 func (c *memCache) autoSyncLoop() { + newe := int64(0) for { - select { - case <-c.stopChan: - return - default: - for { - v := c.eventQueue.PopFront() - if v == nil { - break - } - item := v.(memCacheEvent) - // 添加该key到对应的过期集合中 - // 注意：这里不需要检查存在性， - // 因为在key过期的时候，会和原始的键值对中的过期时间做核对 - newe := c.makeExpireKey(item.e) - c.getOrNewExpireSet(newe).Add(item.k) - // 重新设置对应键名的过期时间 - c.emu.Lock() - c.ekmap[item.k] = newe - c.emu.Unlock() - } - time.Sleep(time.Second) - } + select { + case <-c.stopChan: + return + default: + for { + v := c.eventList.PopFront() + if v == nil { + break + } + item := v.(memCacheEvent) + nowm := gtime.Millisecond() + // 如果用户设置的时间比当前时间还小，那么表示要自动清除了， + // 这里赋值一个当前时间-10秒的时间，在自动清理的goroutine中会自动检测删除该key + if item.e < nowm { + newe = c.makeExpireKey(nowm) - 10000 + } else { + newe = c.makeExpireKey(item.e) + } + // 添加该key到对应的过期集合中 + // 注意：这里不需要检查存在性， + // 因为在key过期的时候，会和原始的键值对中的过期时间做核对。 + c.getOrNewExpireSet(newe).Add(item.k) + // 重新设置对应键名的过期时间 + c.emu.Lock() + c.ekmap[item.k] = newe + c.emu.Unlock() + // LRU(Least Recently Used)操作记录 + if c.cap > 0 { + c.lru.Push(item.k) + } + } + if c.cap > 0 { + // 优先级高的lru key放后面，读取列表 + for { + if v := c.lruGetList.PopFront(); v != nil { + c.lru.Push(v) + } else { + break + } + } + } + time.Sleep(10 * time.Second) + } } } // LRU缓存淘汰处理+自动清理过期键值对 -// 每隔10秒清除过去30秒的键值对数据 +// 每隔10秒清除过去60秒的键值对数据 func (c *memCache) autoClearLoop() { for { select { @@ -303,7 +317,7 @@ func (c *memCache) autoClearLoop() { default: // 缓存过期处理 ek := c.makeExpireKey(gtime.Millisecond()) - eks := []int64{ek - 10000, ek - 20000, ek - 30000} + eks := []int64{ek - 10000, ek - 20000, ek - 30000, ek - 40000, ek - 50000, ek - 60000} for _, v := range eks { if ekset := c.getExpireSet(v); ekset != nil { ekset.Iterator(func(v interface{}) bool { @@ -318,7 +332,7 @@ func (c *memCache) autoClearLoop() { // LRU缓存淘汰清理 if c.cap > 0 { for i := c.Size() - c.cap; i > 0; i-- { - if s := c.lru.Pop(); s != "" { + if s := c.lru.Pop(); s != nil { c.clearByKey(s, true) } } diff --git a/g/os/gcache/gcache_mem_cache_lru.go b/g/os/gcache/gcache_mem_cache_lru.go index abf429cbb..9073a53dd 100644 --- a/g/os/gcache/gcache_mem_cache_lru.go +++ b/g/os/gcache/gcache_mem_cache_lru.go @@ -46,6 +46,11 @@ func (lru *memCacheLru) Remove(key interface{}) { } } +// 当前LRU数据大小 +func (lru *memCacheLru) Size() int { + return lru.data.Size() +} + // 添加LRU数据项 func (lru *memCacheLru) Push(key interface{}) { lru.queue.Push(key) @@ -67,7 +72,7 @@ func (lru *memCacheLru) Print() { } } -// 异步执行协程 +// 异步执行协程，将queue中的数据同步到list中 func (lru *memCacheLru) StartAutoLoop() { for { if v := lru.queue.Pop(); v != nil { diff --git a/g/os/gfsnotify/gfsnotify.go b/g/os/gfsnotify/gfsnotify.go index 9b27d2abf..81c6e743f 100644 --- a/g/os/gfsnotify/gfsnotify.go +++ b/g/os/gfsnotify/gfsnotify.go @@ -52,7 +52,7 @@ func New() (*Watcher, error) { w := &Watcher { watcher : watch, events : gqueue.New(), - closeChan : make(chan struct{}, 1), + closeChan : make(chan struct{}), callbacks : gmap.NewStringInterfaceMap(), } w.startWatchLoop() @@ -83,7 +83,7 @@ func Remove(path string) error { func (w *Watcher) Close() { w.watcher.Close() w.events.Close() - w.closeChan <- struct{}{} + close(w.closeChan) } // 添加对指定文件/目录的监听，并给定回调函数 diff --git a/g/os/grpool/grpool.go b/g/os/grpool/grpool.go index f9ef097de..268d77e9d 100644 --- a/g/os/grpool/grpool.go +++ b/g/os/grpool/grpool.go @@ -28,8 +28,7 @@ type Pool struct { // 该对象与进程同生命周期，无需Close var defaultPool = New() -// 创建goroutine池管理对象，给定过期时间(秒) -// 第二个参数用于限制限制最大的goroutine数量/线程数/worker数量，非必需参数，默认不做限制 +// 创建goroutine池管理对象， 参数用于限制限制最大的goroutine数量/线程数/worker数量，非必需参数，默认不做限制 func New(size...int) *Pool { s := 0 if len(size) > 0 { diff --git a/geg/os/gcache/usage_lru.go b/geg/os/gcache/usage_lru.go index 2ef07ff38..438dea2fb 100644 --- a/geg/os/gcache/usage_lru.go +++ b/geg/os/gcache/usage_lru.go @@ -10,15 +10,18 @@ func main() { // 设置LRU淘汰数量 c := gcache.New(2) - // 添加10个元素 + // 添加10个元素，不过期 for i := 0; i < 10; i++ { c.Set(i, i, 0) } fmt.Println(c.Size()) fmt.Println(c.Keys()) - // 等待一定时间后(默认1秒检查一次)，元素会被按照从旧到新的顺序进行淘汰 - time.Sleep(2*time.Second) + // 读取键名1，保证该键名是优先保留 + fmt.Println(c.Get(1)) + + // 等待一定时间后(默认10秒检查一次)，元素会被按照从旧到新的顺序进行淘汰 + time.Sleep(10*time.Second) fmt.Println(c.Size()) fmt.Println(c.Keys()) } \ No newline at end of file