道：一个用最小4叉堆的 MemoryCache 的 Golang 实现 点击关注公众号，”技术干货”及时达！我是 LEE，老李，一个在 IT 行业摸爬滚打 17 年的技术老兵。 痛点分析最近在做一个基础组件项目新功能开发，需要用到一个内存缓存，这个缓存需要「满足以下几个条件」： 有过期时间，过期后自动自动删除有最大容量，超过容量后不允许再添加新的数据元素的过期时间可以单独设置元素发生变化时，可以触发回调函数回调函数可以单独设置，不需要每个元素都设置代码要实现简单，易于维护性能要好，不要有性能瓶颈代码要有单元测试，保证代码质量需要有 GetOrCreate 方法，可以在缓存中没有的时候，自动创建一个新的元素因为内存缓存在多处被依赖，而且需要创建大量的实例，所以性能要好，而且方便被二次封装，并与开发项目中代码尽可能的融合。 「在写这篇文章之前，我对如下几个项目做了调研：」 ttlcache/v2bigcachefreecachego-cache发现这些项目要不是代码过于复杂，不太好做二次封装或者改造，要不就是性能不太好，要不就是功能不够完善，要不就是没有单元测试，要不就是没有过期时间，要不就是没有最大容量，要不就是没有回调函数，要不就是没有 GetOrCreate 方法。 对于这样方方面面的限制，我只能自己动手，丰衣足食了。正准备大干一场的时候，发现了一个项目，叫做 MemoryCache。 MemoryCache看到这个项目的时候，就觉得这个项目很有意思，激发了我的好奇。随着我对这个项目的深入了解，发现作者写代码的风格很简约、紧凑、直接奔着目标去的，一点都不啰嗦，而且代码的性能也很好，所以我就想把这个项目的代码拿过来，做一些封装，让它更符合我的需求，然后合并进我的项目中。 简介memorycache 是一个基于内存的缓存，通过对 key 完成 hash 运算，然后对 hash 值做位操作(桶的数量一定是 2 的指数个：2^n)，将数据分散到多个桶中，每个桶中的数据都是一个 map + lock + heap，这样就可以实现 O(1) 的时间复杂度来完成数据的增删改查。 结构Key 的 Hash 运算我通过一个简单的例子来说明一下 Key 的 Hash 运算。 a:=13 fmt.Println(a3)//位与运算 fmt.Println(a%3)//取模运算 经常看到有人问，位与运算和取模运算有什么区别，其实这两个运算的结果是一样的，只不过位与运算的效率更高，所以在这里作者选择了位与运算。尤其做为 hash 运算的时候，位与运算的效率更高，所以作者选择了位与运算。(当然社区中有很多开源库中也使用了位与运算，这个算达成共识了) 数据存储桶的实现桶的数量一定是 2 的指数个：2^n，这样就可以通过位与运算来实现 hash 运算，然后将数据分散到多个桶中。这里算是对 Key 的 hash 运算的配合。 那么 bucket 是怎么实现的呢？其实就是一个 map + lock + heap，这样就可以实现 O(1) 的时间复杂度来完成数据的增删改查。假如不考虑 map 中的元素超时的话，直接就用 map + lock 就可以了。但是如果考虑使用 LRU 方式淘汰内部元素的话，在没有 Heap 的方式或者其他的排序方式，map 中的元素超时的话，需要遍历 map，这样就会增加时间复杂度。如果有了 Heap 的方式，就可以实现 O(1) 的时间复杂度来完成对超时的数据剔除。 「说到这里就摆在眼前有几个问题需要解决：」 为什么要使用 Heap 的方式来实现 LRU？什么样的 Heap 才是高效的实现？如何为这些在 Heap 中的元素提供统一的计时器？「为什么要使用 Heap 的方式来实现 LRU？」 Heap 是一种完全二叉树，它分为大根堆和小根堆，大根堆的每个节点的值都大于或等于其左右子节点的值，小根堆的每个节点的值都小于或等于其左右子节点的值。在这里我们使用小根堆来实现 LRU，因为我们需要将最小的元素放在堆顶，这样就可以实现 O(1) 的时间复杂度来完成对超时的数据剔除。 Heap 远比链表的实现方式要高效，因为链表的实现方式，需要遍历链表，这样就会增加时间复杂度。但是 Heap 的实现方式，不需要遍历，只需要对堆顶的元素进行判断，完成对超时的数据剔除。 「什么样的 Heap 才是高效的实现？」 但是传统的 Heap 的实现方式，需要对每个元素都进行排序，这样就会增加操作成本。但是如果采用 4 叉堆的方式，就可以减少排序的次数，这样就可以提高性能。但是 4 叉堆的实现方式，需要对 Heap 的实现方式有一定的了解，这样才能实现高效的 Heap。 目前 4 叉堆的方式，是我在社区看到比较多的项目采用的高效实现。4 叉堆不仅有传统的 Heap 的特点，而且还有一些自己的特点，比如：4 叉堆的每个节点都有 4 个子节点，这样就可以减少排序的次数，提高性能。 如果有更好的实现方式，欢迎大家给我留言。 「如何为这些在 Heap 中的元素提供统一的计时器？」 这个是一个老大难的问题，因为超时的缓存包中，每个缓存的过期时间都不一样，所以需要为每个缓存提供一个单独的计时器，这样才能实现每个缓存的过期时间不一样。但是如果为每个缓存提供一个单独的计时器，就会增加内存的开销，而且也不好管理。 那么就需要设计一个统一的计时器，但是每次更新计时器都是需要创建一个全新的 Timer，这样就会增加 GC 的压力，所以需要设计一个可以复用的 Timer，这样就可以减少 GC 的压力。 充分考虑了在 MemoryCache 中即便缓存了大量的数据，也不会增加太多的 Timer 对象创建和销毁开销，同时对所有的缓存都可以提供统一的计时器，用统一的标尺来衡量所有缓存的对象的生命周期。 快速使用安装gogetgithub.com/lxzan/memorycache 使用举例packagemain import( "fmt" "github.com/lxzan/memorycache" "time" ) funcmain(){ mc:=memorycache.New( memorycache.WithBucketNum(128),//Bucketnumber,recommendedtobeaprimenumber. memorycache.WithBucketSize(1000,10000),//Bucketsize,initialsizeandmaximumcapacity. memorycache.WithInterval(5*time.Second,30*time.Second),//Activecyclecleanupintervalandexpirationtime. ) defermc.Stop() mc.Set("xxx",1,10*time.Second) val,exist:=mc.Get("xxx") fmt.Printf("val=%v,exist=%v\n",val,exist) time.Sleep(32*time.Second) val,exist=mc.Get("xxx") fmt.Printf("val=%v,exist=%v\n",val,exist) } 主要方法这里单独列出来，我在实际项目中觉得比较有用的部分方法。 SetWithCallback这个在往缓存内存放一个对象的时候，可以设置一个回调函数，当缓存中的这个对象发生变化的时候，就会触发这个回调函数。 「回调函数中匹配原因」 const( ReasonExpired=Reason(0)//对象超时 ReasonOverflow=Reason(1)//对象超过最大容量 ReasonDeleted=Reason(2)//对象被删除 ) 「举例：」 mc.SetWithCallback("xxx",1,10*time.Second,func(ele*memorycache.Element,reasonmemorycache.Reason){ ifreason==memorycache.ReasonExpired{ fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) } ifreason==memorycache.ReasonOverflow{ fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) } fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) }) GetOrCreateWithCallback这个在从缓存中获取一个对象的时候，如果缓存中没有这个对象，就会自动创建一个新的对象，然后放入缓存中，同时可以设置一个回调函数，当缓存中的这个对象发生变化的时候，就会触发这个回调函数。 TIPS: 一般缓存的 Get 和 Set 方法都是成对出现的，但是 Get 和 Set 方法所使用的 Lock 中有释放的操作，所以在并发的情况下，可能存在空读情况，所以在这里提供了一个 GetOrCreate 方法，可以在缓存中没有的时候，自动创建一个新的元素。 回调函数的匹配原因，同上。 「举例：」 mc.GetOrCreateWithCallback("xxx",1,10*time.Second,func(ele*memorycache.Element,reasonmemorycache.Reason){ ifreason==memorycache.ReasonExpired{ fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) } ifreason==memorycache.ReasonOverflow{ fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) } fmt.Printf("key=%v,value=%v,reason=%v\n",ele.Key,ele.Value,reason) }) 代码解析这里只解析我在阅读这个项目觉得比较有意思的部分代码，如果有分析不对的地方，欢迎大家指正。 Minimal Quad Heaphttps://github.com/lxzan/memorycache/blob/main/heap.go 在这个 Heap 实现中，Down 方法是一个递归方法，用来将指定的元素下沉到合适的位置，这样就可以保证堆的特性。所以我特意将这个方法单独拿出来，方便大家阅读。 func(c*heap)Down(i,nint){ varindex1=i2+1//i*4+1,i的第一个子节点 ifindex1=n{//如果i的第一个子节点大于等于n，说明i没有子节点，直接返回 return } varindex2=i2+2//i*4+2,i的第二个子节点 varindex3=i2+3//i*4+3,i的第三个子节点 varindex4=i2+4//i*4+4,i的第四个子节点 varj=-1 ifindex4n{//如果i的第四个子节点小于n，说明i有四个子节点 j=c.min(c.min(index1,index2),c.min(index3,index4))//找到i的四个子节点中最小的那个 }elseifindex3n{//如果i的第四个子节点大于等于n，说明i有三个子节点 j=c.min(c.min(index1,index2),index3)//找到i的三个子节点中最小的那个 }elseifindex2n{//如果i的第三个子节点大于等于n，说明i有两个子节点 j=c.min(index1,index2)//找到i的两个子节点中最小的那个 }else{//如果i的第二个子节点大于等于n，说明i只有一个子节点 j=index1//找到i的一个子节点 } ifj=0c.Less(j,i){//如果j大于等于0，说明i有子节点，如果i的子节点小于i，说明i需要下沉 c.Swap(i,j)//交换i和j的位置 c.Down(j,n)//递归调用Down方法，将j下沉到合适的位置 } } 总结MemoryCache 这个库能够提供给我的可能不止这些，但是我目前只用到了这些，所以我只能写出这些。当然 MemoryCache 精巧的代码和高效的性能，还有丰富的单元测试，非常放心将包引入到自己开发的工作当中。不管在使用上还是在二次封装上，都非常的方便。总得来说，MemoryCache 这个库非常的不错，值得阅读和学习、使用。 点击关注公众号，”技术干货”及时达！ 阅读原文