教你打造高性能的 Go 缓存库 - 网站源码_资源分享

译者：luozhiyun 的网志：https://www.luozhiyun.com/archives/531 该文标识符边线： https://github.com/devYun/mycache

我在看许多杰出的开放源码库的这时候看见两个有趣的缓存库 fastcache，在它的如是说主要就有几点特征：

随机存取统计数据要快，即便在mammalian下；即便在数 GB 的缓存中，也要维持较好的操控性，和尽量减少 GC 单次；结构设计尽可能单纯；

责任编辑会透过仿效它写两个单纯的缓存库，进而科学研究其Mach是怎样同时实现这种的最终目标的。期望诸位能略有斩获。

结构设计价值观

在工程项目中，他们时常会加进 Go 缓存库比如说 patrickmn/go-cache库。但许多缓存库只不过都是用两个单纯的 Map 来存放统计数据，那些库在采用的这时候，当mammalian低，信息量少的这时候是没难题的，但在信息量较为大mammalian较为高的这时候会延长 GC 时间，减少缓存重新分配单次。

比如说，他们采用两个单纯的范例：

func main() { a := make(map[string]string, 1e9) for i := 0; i < 10; i++ { runtime.GC() } runtime.KeepAlive(a) }

在那个范例中，预重新分配了大小不一是10亿（1e9) 的 map，接着他们透过 gctrace 输入呵呵 GC 情形：

做试验的自然环境是 Linux，机器配置是 16C 8G ，想更深入细致认知 GC，能看这篇：《 Go 词汇 GC 同时实现基本原理及源标识符预测 https://www.luozhiyun.com/archives/475

》

[root@localhost gotest]# GODEBUG=gctrace=1 go run main.go … gc 6 @13.736s 17%: 0.010+1815+0.004 ms clock, 0.17+0/7254/21744+0.067 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced) gc 7 @15.551s 18%: 0.012+1796+0.005 ms clock, 0.20+0/7184/21537+0.082 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced) gc 8 @17.348s 19%: 0.008+1794+0.004 ms clock, 0.14+0/7176/21512+0.070 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced) gc 9 @19.143s 19%: 0.010+1819+0.005 ms clock, 0.16+0/7275/21745+0.085 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced) gc 10 @20.963s 19%: 0.011+1844+0.004 ms clock, 0.18+0/7373/22057+0.076 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced)

上面展示了最后 5 次 GC 的情形，下面他们看看具体的含义是什么：

gc 1 @0.004s 4%: 0.22+1.4+0.021 ms clock, 1.7+0.009/0.40/0.073+0.16 ms cpu, 4->5->1 MB, 5 MB goal, 8 P gc 10 @20.963s 19%: 0.011+1844+0.004 ms clock, 0.18+0/7373/22057+0.076 ms cpu, 73984->73984->73984 MB, 147968 MB goal, 16 P (forced) gc 10 ：程序启动以来第10次GC @20.963s：距离程序启动到现在的时间 19%：当目前为止，GC 的标记工作所用的CPU时间占总CPU的百分比垃圾回收的时间 0.011 ms：标记开始 STW 时间 1844 ms：mammalian标记时间 0.004 ms：标记终止 STW 时间垃圾回收占用cpu时间 0.18 ms：标记开始 STW 时间 0 ms：mutator assists占用的时间 7373 ms：标记线程占用的时间 22057 ms：idle mark workers占用的时间 0.076 ms：标记终止 STW 时间缓存 73984 MB：标记开始前堆占用大小不一 73984 MB：标记结束后堆占用大小不一 73984 MB：标记完成后存活堆的大小不一 147968 MB goal：标记完成后正在采用的堆缓存的最终目标大小不一 16 P：采用了多少处理器

能从上面的输入看见每次 GC 处理的时间非常的长，占用的 CPU 资源也非常多。那么造成这种的原因是什么呢？

string 实际上底层统计数据结构是由两部分组成，其中包含指向字节数组的指针和数组的大小不一：

type StringHeader struct { Data uintptr Len int }

由于 StringHeader中包含指针，所以每次 GC 的这时候都会扫描每个指针，那么在那个巨大的 map中是包含了非常多的指针的，所以造成了巨大的资源消耗。

在上面的范例 map a 中统计数据大概是这种存储：

两个 map 中里面有多个 bucket ，bucket 里面有两个 bmap 数组用来存放统计数据，但由于 key 和 value 都是 string 类型的，所以在 GC 的这时候还需要根据 StringHeader中的 Data指针扫描 string 统计数据。

对于这种情形，如果所有的 string 字节都在两个单一缓存片段中，他们就能透过偏移来追踪某个字符串在这段缓存中的开始和结束边线。透过追踪偏移，他们不在需要在他们大数组中存储指针，GC 也不在会被困扰。如下：

如同上面所示，如果他们将字符串中的字节统计数据拷贝到两个连续的字节数组 chunks 中，并为那个字节数组提前重新分配好缓存，并且仅存储字符串在数组中的偏移而不是指针。

除了上面所说的优化内容以外，还有其他的方法吗？

只不过他们还能直接从系统 OS 中调用 mmap syscall 进行缓存重新分配，这种 GC 就永远不会对这块缓存进行缓存管理，因此也就不会扫描到它。如下：

func main() { test := “hello syscall” data, _ := syscall.Mmap(-1, 0, 13, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_ANON|syscall.MAP_PRIVATE) p := (*[13]byte)(unsafe.Pointer(&data[0])) for i := 0; i < 13; i++ { p[i] = test[i] } fmt.Println(string(p[:])) }

透过系统调用直接向 OS 申请了 13bytes 的缓存，接着将两个字符串写入到申请的缓存数组中。

所以他们也能透过提前向 OS 申请一块内存，而不是用的这时候才申请缓存，减少频繁的缓存重新分配进而达到提高效能的目的。

源标识符实战

API

他们在开发前先把那个库的 API 定义呵呵：

func New

func New(maxBytes int) *Cache

创建两个 Cache 结构体，传入预设的缓存大小不一，单位是字节。

func (*Cache) Get

func (c *Cache) Get(k []byte) []byte

入的参数是 byte 数组。

func (*Cache) Set

func (c *Cache) Set(k, v []byte)

设置键值对到缓存中，k 是键，v 是值，参数都是 byte 数组。

结构体

const bucketsCount = 512 type Cache struct { buckets [bucketsCount]bucket } type bucket struct { // 随机存取锁 mu sync.RWMutex // 二维数组，存放统计数据的地方，是两个环形链表 chunks [][]byte // 索引字典 m map[uint64]uint64 // 索引值 idx uint64 // chunks 被重写的单次，用来校验环形链表中统计数据有效性 gen uint64 }

透过他们上面的预测，能看见，实际上真正存放统计数据的地方是 chunks 二维数组，在同时实现上是透过 m 字段来映射索引路径，根据 chunks 和 gen 两个字段来构建两个环形链表，环形链表每转一圈 gen 就会加一。

初始化

func New(maxBytes int) *Cache { if maxBytes <= 0 { panic(fmt.Errorf(“maxBytes must be greater than 0; got %d”, maxBytes)) } var c Cache // 算出每个桶的大小不一 maxBucketBytes := uint64((maxBytes + bucketsCount – 1) / bucketsCount) for i := range c.buckets[:] { // 对桶进行初始化 c.buckets[i].Init(maxBucketBytes) } return &c }

他们会设置两个 New 函数来初始化他们 Cache 结构体，在 Cache 结构体中会将缓存的统计数据大小不一平均重新分配到每个桶中，接着对每个桶进行初始化。

const bucketSizeBits = 40 const maxBucketSize uint64 = 1 << bucketSizeBits const chunkSize = 64 * 1024 func (b *bucket) Init(maxBytes uint64) { if maxBytes == 0 { panic(fmt.Errorf(“maxBytes cannot be zero”)) } // 他们这里限制每个桶最大的大小不一是 1024 GB if maxBytes >= maxBucketSize { panic(fmt.Errorf(“too big maxBytes=%d; should be smaller than %d”, maxBytes, maxBucketSize)) } // 初始化 Chunks 中每个 Chunk 大小不一为 64 KB，计算 chunk 数量 maxChunks := (maxBytes + chunkSize – 1) / chunkSize b.chunks = make([][]byte, maxChunks) b.m = make(map[uint64]uint64) // 初始化 bucket 结构体 b.Reset() }

在这里会将桶里面的缓存按 chunk 进行重新分配，每个 chunk 占用缓存约为 64 KB。在最后会调用 bucket 的 Reset 方法对 bucket 结构体进行初始化。

func (b *bucket) Reset() { b.mu.Lock() chunks := b.chunks // 遍历 chunks for i := range chunks { // 将 chunk 中的缓存归还到缓存中 putChunk(chunks[i]) chunks[i] = nil } // 删除索引字典中所有的统计数据 bm := b.m for k := range bm { delete(bm, k) } b.idx = 0 b.gen = 1 b.mu.Unlock() }

Reset 方法十分单纯，主要就就是清空 chunks 数组、删除索引字典中所有的统计数据和重置索引 idx 和 gen 的值。

在上面这个方法中有两个 putChunk ，只不过那个就是直接操作他们提前向 OS 申请好的缓存，相应的还有两个 getChunk 方法。下面他们具体看看 Chunk 的操作。

Chunk 操作

getChunk

const chunksPerAlloc = 1024 const chunkSize = 64 * 1024 var ( freeChunks []*[chunkSize]byte freeChunksLock sync.Mutex ) func getChunk() []byte { freeChunksLock.Lock() if len(freeChunks) == 0 { // 重新分配 64 * 1024 * 1024 = 64 MB 缓存 data, err := syscall.Mmap(-1, 0, chunkSize*chunksPerAlloc, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_ANON|syscall.MAP_PRIVATE) if err != nil { panic(fmt.Errorf(“cannot allocate %d bytes via mmap: %s”, chunkSize*chunksPerAlloc, err)) } // 循环遍历 data 统计数据 for len(data) > 0 { //将从系统重新分配的缓存分为 64 * 1024 = 64 KB 大小不一，存放到 freeChunks中 p := (*[chunkSize]byte)(unsafe.Pointer(&data[0])) freeChunks = append(freeChunks, p) data – 1 p := freeChunks[n] freeChunks[n] = nil freeChunks = freeChunks[:n] freeChunksLock.Unlock() return p[:] }

初次调用 getChunk 函数时会采用系统调用重新分配 64MB 的缓存，接着循环将缓存切将要如是说到的 Cache 的 set 方法中采加进，所以需要考虑到mammalian难题，所以在这里加了锁。

putChunk

func putChunk(chunk []byte) { if chunk == nil { return } chunk = chunk[:chunkSize] p := (*[chunkSize]byte)(unsafe.Pointer(&chunk[0])) freeChunksLock.Lock() freeChunks = append(freeChunks, p) freeChunksLock.Unlock() }

putChunk 函数就是将缓存统计数据还回到 freeChunks 空闲列表中，会在 bucket 的 Reset 方法中被调用。

Set

const bucketsCount = 512 func (c *Cache) Set(k, v []byte) { h := xxhash.Sum64(k) idx := h % bucketsCount c.buckets[idx].Set(k, v, h) }

Set 方法里面会根据 k 的值做两个 hash，接着取模映射到 buckets 桶中，这里用的 hash 库是 cespare/xxhash。

最主要就的还是 buckets 里面的 Set 方法：

func (b *bucket) Set(k, v []byte, h uint64) { // 限定 k v 大小不一不能超过 2bytes if len(k) >= (1<<16) || len(v) >= (1<<16) { return } // 4个byte 设置每条统计数据的统计数据头 var kvLenBuf [4]byte kvLenBuf[0] = byte(uint16(len(k)) >> 8) kvLenBuf[1] = byte(len(k)) kvLenBuf[2] = byte(uint16(len(v)) >> 8) kvLenBuf[3] = byte(len(v)) kvLen := uint64(len(kvLenBuf) + len(k) + len(v)) // 校验呵呵大小不一 if kvLen >= chunkSize { return } b.mu.Lock() // 当前索引边线 idx := b.idx // 存放完统计数据后索引的边线 idxNew := idx + kvLen // 根据索引找到在 chunks 的边线 chunkIdx := idx / chunkSize chunkIdxNew := idxNew / chunkSize // 新的索引是否超过当前索引 // 因为还有chunkIdx等于chunkIdxNew情形，所以需要先判断呵呵 if chunkIdxNew > chunkIdx { // 校验是否新索引已到chunks数组的边界 // 已到边界，那么循环链表从头开始 if chunkIdxNew >= uint64(len(b.chunks)) { idx = 0 idxNew = kvLen chunkIdx = 0 b.gen++ // 当 gen 等于 1<<genSizeBits时，才会等于0 // 也就是用来限定 gen 的边界为1<<genSizeBits if b.gen&((1<<genSizeBits)-1) == 0 { b.gen++ } } else { // 未到 chunks数组的边界,从下两个chunk开始 idx = chunkIdxNew * chunkSize idxNew = idx + kvLen chunkIdx = chunkIdxNew } // 重置 chunks[chunkIdx] b.chunks[chunkIdx] = b.chunks[chunkIdx][:0] } chunk := b.chunks[chunkIdx] if chunk == nil { chunk = getChunk() // 清空切片 chunk = chunk[:0] } // 将统计数据 append 到 chunk 中 chunk = append(chunk, kvLenBuf[:]…) chunk = append(chunk, k…) chunk = append(chunk, v…) b.chunks[chunkIdx] = chunk // 因为 idx 不能超过bucketSizeBits，所以用两个 uint64 同时表示gen和idx // 所以高于bucketSizeBits边线表示gen // 低于bucketSizeBits边线表示idx b.m[h] = idx | (b.gen << bucketSizeBits) b.idx = idxNew b.mu.Unlock() }

在这段标识符开头实际上我会限制键值的大小不一不能超过 2bytes；接着将 2bytes 大小不一长度的键值封装到 4bytes 的 kvLenBuf 作为统计数据头，统计数据头和键值的总长度是不能超过两个 chunk 长度，也就是 64 * 1024；接着计算出原索引 chunkIdx 和新索引 chunkIdxNew，用来判断这次添加的统计数据加上原来的统计数据有没超过两个 chunk 长度；根据新的索引找到对应的 chunks 中的边线，接着将键值和 kvLenBuf 追加到 chunk 后面；设置新的 idx 和 m 字典对应的值，m 字典中存放的是 gen 和 idx 透过取与的放置存放。

在 Set 两个键值对会有 4bytes 的 kvLenBuf 作为统计数据头，后面的统计数据会接着 key 和 value ，在 kvLenBuf 中，前两个 byte 分别代表了 key 长度的低位和高位；后两个 byte 分别代表了 value 长度的低位和高位，统计数据图大致如下：

下面举个范例来看看是是怎样利用 chunks 那个二维数组来同时实现环形链表的。

他们在 bucket 的 Init 方法中会根据传入 maxBytes 桶字节数来设置 chunks 的长度大小不一，由于每个 chunk 大小不一都是 64 * 1024bytes，那么他们设置 3 * 64 * 1024bytes 大小不一的桶，那么 chunks 数组长度就为 3。

如果当前算出 chunkIdx 在 chunks 数组为 1 的边线，并且在 chunks[1] 的边线中，还剩下 6bytes 未被采用，那么有如下几种情形：

现在假设放入的键值长度都是 1byte，那么在 chunks[1] 的边线中剩下的 6bytes 刚好能放下；

现在假设放入的键值长度超过了 1byte，那么在 chunks[1] 的边线中剩下的边线就放不下，只能放入到 chunks[2] 的边线中。

如果当前算出 chunkIdx 在 chunks 数组为 2 的边线，并且现在 Set 两个键值，经过计算 chunkIdxNew 为 3，已经超过了 chunks 数组长度，那么会将索引重置，重新将统计数据从 chunks[0] 开始放置，并将 gen 加一，表示已经跑完一圈了。

Get

func (c *Cache) Get(dst, k []byte) []byte { h := xxhash.Sum64(k) idx := h % bucketsCount dst, _ = c.buckets[idx].Get(dst, k, h, true) return dst }

这里和 Set 方法是一样的，首先是要找到对应的桶的边线，接着才去桶里面拿统计数据。需要注意的是，这里的 dst 能从外部传入两个切片，以达到减少重复重新分配返回值。

func (b *bucket) Get(dst, k []byte, h uint64,returnDst bool) ([]byte, bool) { found := false b.mu.RLock() v := b.m[h] bGen := b.gen & ((1 << genSizeBits) – 1) if v > 0 { // 高于bucketSizeBits边线表示gen gen := v >> bucketSizeBits // 低于bucketSizeBits边线表示idx idx := v & ((1 << bucketSizeBits) – 1) // 这里说明chunks还没被写满 if gen == bGen && idx < b.idx || // 这里说明chunks已被写满，并且当前统计数据没被覆盖 gen+1 == bGen && idx >= b.idx || // 这里是边界条件gen已是最大，并且chunks已被写满bGen从1开始，，并且当前统计数据没被覆盖 gen == maxGen && bGen == 1 && idx >= b.idx { chunkIdx := idx / chunkSize // chunk 索引边线不能超过 chunks 数组长度 if chunkIdx >= uint64(len(b.chunks)) { goto end } // 找到统计数据所在的 chunk chunk := b.chunks[chunkIdx] // 透过取模找到该key 对应的统计数据在 chunk 中的边线 idx %= chunkSize if idx+4 >= chunkSize { goto end } // 前 4bytes 是统计数据头 kvLenBuf := chunk[idx : idx+4] // 透过统计数据头算出键值的长度 keyLen := (uint64(kvLenBuf[0]) << 8) | uint64(kvLenBuf[1]) valLen := (uint64(kvLenBuf[2]) << 8) | uint64(kvLenBuf[3]) idx += 4 if idx+keyLen+valLen >= chunkSize { goto end } // 如果键值是一致的，表示找到该统计数据 if string(k) == string(chunk[idx:idx+keyLen]) { idx += keyLen // 返回该键对应的值 if returnDst { dst = append(dst, chunk[idx:idx+valLen]…) } found = true } } } end: b.mu.RUnlock() return dst, found }

找到 gen 和 idx 索引之后就是边界条件的判断了，用两个 if 条件来进行判断：

gen == bGen && idx < b.idx

这里是判断如果是在环形链表的同一次循环中，那么 key 对应的索引应该小于当前桶的索引；

gen+1 == bGen && idx >= b.idx

这里表示当前桶已经进入到下两个循环中，所以需要判断 key 对应的索引是不是大于当前索引，以表示当前 key 对应的值没被覆盖；

gen == maxGen && bGen == 1 && idx >= b.idx

因为 gen 和 idx 索引要塞到 uint64 类型的字段中，所以留给 gen 的最大值只有 maxGen = 1<< 24 -1，超过了 maxGen 会让 gen 从 1 开始。所以这里如果 key 对应 gen 等于 maxGen ，那么当前的 bGen 应该等于 1，并且 key 对应的索引还应该大于当前 idx，这种才那个键值对才不会被覆盖。

判断完边界条件之后就会找到对应的 chunk ，接着取模后找到统计数据边线，透过偏移量找到并取出值。

Benchmark

下面我上呵呵过后的 Benchmark：

标识符边线： https://github.com/devYun/mycache/blob/main/cache_timing_test.go

GOMAXPROCS=4 go test -bench=Set|Get -benchtime=10s goos: linux goarch: amd64 pkg: gotest // GoCache BenchmarkGoCacheSet-4 836 14595822 ns/op 4.49 MB/s 2167340 B/op 65576 allocs/op BenchmarkGoCacheGet-4 3093 3619730 ns/op 18.11 MB/s 5194 B/op 23 allocs/op BenchmarkGoCacheSetGet-4 236 54379268 ns/op 2.41 MB/s 2345868 B/op 65679 allocs/op // BigCache BenchmarkBigCacheSet-4 1393 12763995 ns/op 5.13 MB/s 6691115 B/op 8 allocs/op BenchmarkBigCacheGet-4 2526 4342561 ns/op 15.09 MB/s 650870 B/op 131074 allocs/op BenchmarkBigCacheSetGet-4 1063 11180201 ns/op 11.72 MB/s 4778699 B/op 131081 allocs/op // standard map BenchmarkStdMapSet-4 1484 7299296 ns/op 8.98 MB/s 270603 B/op 65537 allocs/op BenchmarkStdMapGet-4 4278 2480523 ns/op 26.42 MB/s 2998 B/op 15 allocs/op BenchmarkStdMapSetGet-4 343 39367319 ns/op 3.33 MB/s 298764 B/op 65543 allocs/op // sync.map BenchmarkSyncMapSet-4 756 15951363 ns/op 4.11 MB/s 3420214 B/op 262320 allocs/op BenchmarkSyncMapGet-4 11826 1010283 ns/op 64.87 MB/s 1075 B/op 33 allocs/op BenchmarkSyncMapSetGet-4 1910 5507036 ns/op 23.80 MB/s 3412764 B/op 262213 allocs/op PASS ok gotest 215.182s

上面的测试是 GoCache、BigCache、Map、sync.Map 的情形。下面是本篇文章中所开发的缓存库的测试：

// myCachce BenchmarkCacheSet-4 4371 2723208 ns/op 24.07 MB/s 1306 B/op 2 allocs/op BenchmarkCacheGet-4 6003 1884611 ns/op 34.77 MB/s 951 B/op 1 allocs/op BenchmarkCacheSetGet-4 2044 6611759 ns/op 19.82 MB/s 2797 B/op 5 allocs/op

能看见缓存重新分配是几乎就不存在，操作速度在上面的库中也是佼佼者的存在。

总结

在本该文根据其他缓存库，并预测了如果用 Map 作为缓存所存在的难题，接着引出存在那个难题的原因，并提出解决方案；在他们的缓存库中，第一是透过采用索引加缓存块的方式来存放缓存统计数据，再来是透过 OS 系统调用来进行缓存重新分配让他们的缓存统计数据块脱离了 GC 的控制，进而做到降低 GC 频率提高mammalian的目的。

只不过不只是缓存库，在他们的工程项目中当遇到需要采用大量的带指针的统计数据结构并需要长时间维持引用的这时候，也是需要注意这种做可能会引发 GC 难题，进而给系统带来隐患。