在低延迟，高并发的系统中，不可避免的会用到本地内存作为缓存，FreeCache 就是使用golang实现的本地缓存系统，良好的特性使得它目前用在我们的生产环境中。一直以来都对他的实现很感兴趣，这次通过分析源码的方式，学习一下。
项目地址及特性
项目地址
特性:
- 存储数以亿计的条目
- 零 GC 开销
- 高并发线程安全访问
- 纯 Go 实现
- 支持键值过期
- 近似 LRU 的淘汰算法
- 严格限制内存使用
- 迭代器支持
内部数据结构
freecache 是如何做到 0 GC 和高并发、低延时的呢，先看一下他的主要结构
const (
segmentCount = 256
segmentAndOpVal = 255
)
type Cache struct {
locks [segmentCount]sync.Mutex
segments [segmentCount]segment
}
// 每个分片的数据结构
type segment struct {
rb RingBuf // 环形数组
segId int // hashVal & 255 后的id
hitCount int64
missCount int64
entryCount int64
totalCount int64 // 环形数组 ring buffer 中的数据的总条数
totalTime int64 // 用于计算 lru
totalEvacuate int64
totalExpired int64
overwrites int64
// 环形数组可用容量，用于维护环形数组，保证写入新数据而不会覆盖旧数据
vacuumLen int64
// entry 索引每个 slot 的长度的数组。当 slotId 冲突时，对应 slot 的值会+1
slotLens [256]int32
// entry 索引每个 slot 的容量
//只要有一个 slot 的长度等于 slotCap 时，就会触发扩容
slotCap int32
// 存储 entry 索引的切片，按照 hash16 顺序排列，256个 slot 共用
slotsData []entryPtr
}
type RingBuf struct {
begin int64 // 环形数据的起始位置
end int64 // 环形数据的结束位置
data []byte // 存储所有的 entry，容量是固定值
index int
} 光看结构好像不能一目了然，撸个数据结构图
freecache 数据结构图
通过结构图，可以看出 freecache 是将缓存空间划分为 256 个 segment，每个 segment 都有相同都存储空间，并有一把锁。
每个 segment 包含 256 个索引槽，用来对 key 进行快速检索，通过 uint8(hashVal >> 8) 运算获取到 key 对应索引槽的 slotId。每个槽中又有 n 个索引，每个槽中的索引数量是统一的，由 slotCap 进行控制，当某个槽中的索引的数量大于 slotCap 时，就会触发整个索引的扩容。每个槽的索引数据是存储在 slotsData 这个索引切片中的，256 个槽共用同一个索引切片，每个槽在索引切片中都是按照 hash16 顺序排列的（如结构图显示，同一颜色的 solt 对应同一颜色的 ptr 切片）。根据 slotId 和 slotCap ，可以轻松定位到某个槽下所有的索引切片，提高 key 检索效率。
每个 segment 包含1个环形数组 RingBuf ，用来存储实际的数据。在 freecache 中，将每个 k/v 数据定义为一个 entry ，缓存中有多少个 key ，就有多少个 entry。
为什么可以高并发线程安全访问
当对 key 进行 set、get、del 等操作时，freecache 使用 xxhash 这个 hash 方法，对 key 计算得到一个64位的 hashValue。通过 hashVal & 255 得到 segId，将 key 定位到具体的 segment，并对 segment 加锁。由于每次只对单个 segment 加锁，不同 segment 之间可以并发进行 key 操作。
// 以 set 为例，get、del 等其他操作都是相似的
func (cache *Cache) Set(key, value []byte, expireSeconds int) (err error) {
hashVal := hashFunc(key)
segID := hashVal & segmentAndOpVal
cache.locks[segID].Lock()
err = cache.segments[segID].set(key, value, hashVal, expireSeconds)
cache.locks[segID].Unlock()
return
} 为什么是零 GC 的开销
通过结构图和源码，可以看出 freecache 的底层只有 512 个指针(256 个 segment ，每个 segment 包含一个 slotsData 切片和一个 RingBuf)，所以 freecache 的对GC开销几乎为0。
set 操作的具体流程
话不多说，先上图
freecache set 流程图
freecache set 时，先查询当前 key 是否已存在，根据 segId 和 slotId 快速定位到 entry 所在的索引槽。由于索引槽中的索引都是按照 hash16 顺序排列的，可以用二分查找法检索（复杂度 O(log2n)）。
/**
* 判断 key 是否在索引切片中存在
*/slotId := uint8(hashVal >> 8)
hash16 := uint16(hashVal >> 16)
// 根据 slotId 和 slotCap
// 从公用的 slotsData 切片中获取第 slotId 个槽所对应的索引切片
slot := seg.getSlot(slotId)
// 在槽对应的索引中，搜索 key 是否存在
idx, match := seg.lookup(slot, hash16, key)
func (seg *segment) getSlot(slotId uint8) []entryPtr {
slotOff := int32(slotId) * seg.slotCap
return seg.slotsData[slotOff : slotOff+seg.slotLens[slotId] : slotOff+seg.slotCap]
}
func (seg *segment) lookup(slot []entryPtr, hash16 uint16, key []byte) (idx int, match bool) {
// 二分查找法获取到第一个符合条件的 hash16
idx = entryPtrIdx(slot, hash16)
for idx < len(slot) {
ptr := &slot[idx]
if ptr.hash16 != hash16 {
break
}
// 相同的 hash16 ，需要到 RingBuf 比对 key 是否相同
match = int(ptr.keyLen) == len(key) && seg.rb.EqualAt(key, ptr.offset+ENTRY_HDR_SIZE)
if match {
return
}
idx++
}
return
}
func entryPtrIdx(slot []entryPtr, hash16 uint16) (idx int) {
high := len(slot)
for idx < high {
mid := (idx + high) >> 1
oldEntry := &slot[mid]
if oldEntry.hash16 < hash16 {
idx = mid + 1
} else {
high = mid
}
}
return
}
简单的画了个图，假设根据 slotId 和 slotCap 获取到 freecache 数据结构图中紫色的索引切片
freecache 索引检索图
当 key 对应的索引存在时，需要查看 RingBuf 中， entry 以前的容量是多少。如果 entry 容量大于 set 的 value 长度，直接更新原来 entry 的头部和 value (复杂度O(1))。如果 entry 容量不充足的话，为了避免移动 RingBuf 数组的数据，不直接对原来的 entry 进行扩容，而是将原来的 entry 标记为删除（不直接删除，迁移数据，只标记，当 RingBuf 空间不够需要淘汰数据的时候会真正删除），然后在 RingBuf 的环尾追加新的 entry(复杂度O(1)，新 entry 容量如下)。
// 每次扩容原空间的两倍，直到满足插入 value 的空间为止
for hdr.valCap < hdr.valLen {
hdr.valCap *= 2
} 当 key 对应的索引不存在时，要先将索引添加到索引切片中，由于索引是有序的，在插入索引时，可能会对老的索引数据进行迁移(复杂度O(n))，如果索引切片容量不够，还会触发索引切片扩容。索引添加成功后，再将 entry 追加到 RingBuf 的环尾(复杂度O(1))。
/**
* 将索引插入到索引切片
*/func (seg *segment) insertEntryPtr(
slotId uint8, hash16 uint16, offset int64, idx int, keyLen uint16,
) {
// 只要有一个索引槽的容量大于 slotCap
// 就会对整个索引切片进行扩容
// 扩容后容量是扩容前的两倍
if seg.slotLens[slotId] == seg.slotCap {
seg.expand()
}
seg.slotLens[slotId]++
atomic.AddInt64(&seg.entryCount, 1)
slot := seg.getSlot(slotId)
copy(slot[idx+1:], slot[idx:])
slot[idx].offset = offset
slot[idx].hash16 = hash16
slot[idx].keyLen = keyLen
}
func (seg *segment) expand() {
newSlotData := make([]entryPtr, seg.slotCap*2*256)
for i := 0; i < 256; i++ {
off := int32(i) * seg.slotCap
copy(newSlotData[off*2:], seg.slotsData[off:off+seg.slotLens[i]])
}
seg.slotCap *= 2
seg.slotsData = newSlotData
} 以 freecache 数据结构图中紫色的索引切片为例，当插入索引，不需要扩容时
freecache 索引插入不需要扩容图
以 freecache 数据结构图中黄色的索引切片为例当插入索引，需要扩容时
freecache 索引插入扩容前图
freecache 索引插入扩容后图
如果将 entry 追加到 RingBuf 的环尾时，RingBuf 容量不够时，freecache 会根据策略淘汰数据，释放出容量。
近似 LRU 的淘汰流程
freecache lru 图
为什么是近似 LRU 的淘汰呢，先看下源码
leastRecentUsed := int64(oldHdr.accessTime)*atomic.LoadInt64(&seg.totalCount) <= atomic.LoadInt64(&seg.totalTime)
将上边等式变换一下
//oldHdr.accessTime: RingBuf 中第一个 entry 数据最近一次访问的时间戳
//seg.totalCount: RingBuf 中 entry 的总数，包括过期和标记删除的 entry
//seg.totalTime: RingBuf 中每个 entry 最近一次访问的时间戳总和
leastRecentUsed := int64(oldHdr.accessTime) <= atomic.LoadInt64(&seg.totalTime)/atomic.LoadInt64(&seg.totalCount)
atomic.LoadInt64(&seg.totalTime)/atomic.LoadInt64(&seg.totalCount) 表示 RingBuf 所有 entry 最近一次访问时间戳的平均值。 freecach 认为当某个 entry 的 accessTime 小于等于这个平均值，则该 entry 是可以被淘汰的。
freecache 选择将 accessTime 小于等于平均 accessTime 的 entry 淘汰，不是严格意义上的 LRU 算法，但是确实会将最近较少使用的 entry 淘汰掉，所以是一种近 LRU 的算法。 当然，此算法也有可能会将较新的 entry 数据淘汰掉。当 RingBuf 头部数据都比较新，并且没有过期时，会导致一直找不到符合条件的 entry 进行淘汰，这样就无法空出足够的空间让新数据写入，程序会无限循环的将数据从头部迁移到尾部导致 cpu 被耗尽。
为了防止这样的意外发生，也为了保证 set 操作的性能，当连续第6次 RingBuf 的第一个 entry 还不符合淘汰条件时，该 entry 会被强制淘汰。
key 过期后缓存数据的清除策略
freecache 不会主动清除过期的数据(包括索引和 entry 数据)。当数据过期后，在被标记删除之前，key 被重新 set 进来，如果 entry 的容量充足，是可以进行复用的。当数据过期后，当 get/touch 操作或 LRU 的时候，会将 key 对应的索引删除，entry 不会被直接删除，只会被标记为删除状态，等到 LRU 的时候，才会将 entry 删除，为 RingBuf 腾出空间。
