Golang中的缓存库freecache怎么用？

go开发缓存场景一般使用map或者缓存框架，为了线程安全会使用sync.Map或线程安全的缓存框架。

缓存场景中如果数据量大于百万级别，需要特别考虑数据类型对于gc的影响（注意string类型底层是指针+Len+Cap，因此也算是指针类型），如果缓存key和value都是非指针类型的话就无需多虑了。

但实际应用场景中，key和value是(包含)指针类型数据是很常见的，因此使用缓存框架需要特别注意其对gc影响，从是否对GC影响角度来看缓存框架大致分为2类：

零GC开销：比如freecache或bigcache这种，底层基于ringbuf，减小指针个数；

有GC开销：直接基于Map来实现的缓存框架。

对于map而言，gc时会扫描所有key/value键值对，如果其都是基本类型，那么gc便不会再扫描。

下面以freecache为例分析下其实现原理，代码示例如下：

func main() {

  cacheSize := 100 * 1024 * 1024

  cache := freecache.NewCache(cacheSize)

  fori := 0; i < N; i++ {

     str := strconv.Itoa(i)

     _ = cache.Set([]byte(str), []byte(str), 1)

  }

  now := time.Now()

  runtime.GC()

  fmt.Printf("freecache, GC took: %s\n", time.Since(now))

  _, _ = cache.Get([]byte("aa"))

  now = time.Now()

  fori := 0; i < N; i++ {

     str := strconv.Itoa(i)

     _, _ = cache.Get([]byte(str))

  }

  fmt.Printf("freecache, Get took: %s\n\n", time.Since(now))

}

1 初始化

freecache.NewCache会初始化本地缓存，size表示存储空间大小，freecache会初始化256个segment，每个segment是独立的存储单元，freecache加锁维度也是基于segment的，每个segment有一个ringbuf，初始大小为size/256。freecache号称零GC的来源就是其指针是固定的，只有512个，每个segment有2个，分别是rb和slotData（注意切片为指针类型）。

type segment struct {

  rb            RingBuf // ring buffer that stores data

  segId         int

  _             uint32  // 占位

  missCount     int64

  hitCount      int64

  entryCount    int64

  totalCount    int64      // number of entries in ring buffer, including deleted entries.

  totalTime     int64      // used to calculate least recent used entry.

  timer         Timer      // Timer giving current time

  totalEvacuate int64      // used for debug

  totalExpired  int64      // used for debug

  overwrites    int64      // used for debug

  touched       int64      // used for debug

  vacuumLen     int64      // up to vacuumLen, new data can be written without overwriting old data.

  slotLens      [256]int32 // The actual length for every slot.

  slotCap       int32      // max number of entry pointers a slot can hold.

  slotsData     []entryPtr // 索引指针

}

func NewCacheCustomTimer(size int, timer Timer) (cache *Cache) {

   cache = new(Cache)

   fori := 0; i < segmentCount; i++ {

       cache.segments[i] = newSegment(size/segmentCount, i, timer)

   }

}

func newSegment(bufSize int, segId int, timer Timer) (seg segment) {

   seg.rb = NewRingBuf(bufSize, 0)

   seg.segId = segId

   seg.timer = timer

   seg.vacuumLen = int64(bufSize)

   seg.slotCap = 1

   seg.slotsData = make([]entryPtr, 256*seg.slotCap) // 每个slotData初始化256个单位大小

}

2 读写流程

freecache的key和value都是[]byte数组，使用时需要自行序列化和反序列化，如果缓存复杂对象不可忽略其序列化和反序列化带来的影响，首先看下Set流程：

1

_ = cache.Set([]byte(str), []byte(str), 1)

Set流程首先对key进行hash，hashVal类型uint64，其低8位segID对应segment数组，低8-15位表示slotId对应slotsData下标，高16位表示slotsData下标对应的[]entryPtr某个数据，这里需要查找操作。注意[]entryPtr数组大小为slotCap(初始为1)，当扩容时会slotCap倍增。

每个segment对应一个lock（sync.Mutex），因此其能够支持较大并发量，而不像sync.Map只有一个锁。

func (cache *Cache) Set(key, value []byte, expireSeconds int) (err error) {

  hashVal := hashFunc(key)

  segID := hashVal & segmentAndOpVal // 低8位

  cache.locks[segID].Lock() // 加锁

  err = cache.segments[segID].set(key, value, hashVal, expireSeconds)

  cache.locks[segID].Unlock()

}

func (seg *segment) set(key, value []byte, hashVal uint64, expireSeconds int) (err error) {

  slotId := uint8(hashVal >> 8)

  hash16 := uint16(hashVal >> 16)

  slot := seg.getSlot(slotId)

  idx, match := seg.lookup(slot, hash16, key)

  varhdrBuf [ENTRY_HDR_SIZE]byte

  hdr := (*entryHdr)(unsafe.Pointer(&hdrBuf[0]))

  ifmatch { // 有数据更新操作

     matchedPtr := &slot[idx]

     seg.rb.ReadAt(hdrBuf[:], matchedPtr.offset)

     hdr.slotId = slotId

     hdr.hash16 = hash16

     hdr.keyLen = uint16(len(key))

     originAccessTime := hdr.accessTime

     hdr.accessTime = now

     hdr.expireAt = expireAt

     hdr.valLen = uint32(len(value))

     ifhdr.valCap >= hdr.valLen {

        // 已存在数据value空间能存下此次value大小

        atomic.AddInt64(&seg.totalTime, int64(hdr.accessTime)-int64(originAccessTime))

        seg.rb.WriteAt(hdrBuf[:], matchedPtr.offset)

        seg.rb.WriteAt(value, matchedPtr.offset+ENTRY_HDR_SIZE+int64(hdr.keyLen))

        atomic.AddInt64(&seg.overwrites, 1)

        return

     }

     // 删除对应entryPtr，涉及到slotsData内存copy，ringbug中只是标记删除

     seg.delEntryPtr(slotId, slot, idx)

     match = false

     // increase capacity and limit entry len.

     forhdr.valCap < hdr.valLen {

        hdr.valCap *= 2

     }

     ifhdr.valCap > uint32(maxKeyValLen-len(key)) {

        hdr.valCap = uint32(maxKeyValLen - len(key))

     }

  } else{ // 无数据

     hdr.slotId = slotId

     hdr.hash16 = hash16

     hdr.keyLen = uint16(len(key))

     hdr.accessTime = now

     hdr.expireAt = expireAt

     hdr.valLen = uint32(len(value))

     hdr.valCap = uint32(len(value))

     ifhdr.valCap == 0 { // avoid infinite loop when increasing capacity.

        hdr.valCap = 1

     }

  }

   

  // 数据实际长度为 ENTRY_HDR_SIZE=24 + key和value的长度    

  entryLen := ENTRY_HDR_SIZE + int64(len(key)) + int64(hdr.valCap)

  slotModified := seg.evacuate(entryLen, slotId, now)

  ifslotModified {

     // the slot has been modified during evacuation, we need to looked up for the 'idx' again.

     // otherwise there would be index out of bound error.

     slot = seg.getSlot(slotId)

     idx, match = seg.lookup(slot, hash16, key)

     // assert(match == false)

  }

  newOff := seg.rb.End()

  seg.insertEntryPtr(slotId, hash16, newOff, idx, hdr.keyLen)

  seg.rb.Write(hdrBuf[:])

  seg.rb.Write(key)

  seg.rb.Write(value)

  seg.rb.Skip(int64(hdr.valCap - hdr.valLen))

  atomic.AddInt64(&seg.totalTime, int64(now))

  atomic.AddInt64(&seg.totalCount, 1)

  seg.vacuumLen -= entryLen

  return

}

seg.evacuate会评估ringbuf是否有足够空间存储key/value，如果空间不够，其会从空闲空间尾部后一位（也就是待淘汰数据的开始位置）开始扫描（oldOff := seg.rb.End() + seg.vacuumLen - seg.rb.Size()），如果对应数据已被逻辑deleted或者已过期，那么该块内存可以直接回收，如果不满足回收条件，则将entry从环头调换到环尾，再更新entry的索引，如果这样循环5次还是不行，那么需要将当前oldHdrBuf回收以满足内存需要。

执行完seg.evacuate所需空间肯定是能满足的，然后就是写入索引和数据了，insertEntryPtr就是写入索引操作，当[]entryPtr中元素个数大于seg.slotCap(初始1)时，需要扩容操作，对应方法见seg.expand，这里不再赘述。

写入ringbuf就是执行rb.Write即可。

func (seg *segment) evacuate(entryLen int64, slotId uint8, now uint32) (slotModified bool) {

  varoldHdrBuf [ENTRY_HDR_SIZE]byte

  consecutiveEvacuate := 0

  forseg.vacuumLen < entryLen {

     oldOff := seg.rb.End() + seg.vacuumLen - seg.rb.Size()

     seg.rb.ReadAt(oldHdrBuf[:], oldOff)

     oldHdr := (*entryHdr)(unsafe.Pointer(&oldHdrBuf[0]))

     oldEntryLen := ENTRY_HDR_SIZE + int64(oldHdr.keyLen) + int64(oldHdr.valCap)

     ifoldHdr.deleted { // 已删除

        consecutiveEvacuate = 0

        atomic.AddInt64(&seg.totalTime, -int64(oldHdr.accessTime))

        atomic.AddInt64(&seg.totalCount, -1)

        seg.vacuumLen += oldEntryLen

        continue

     }

     expired := oldHdr.expireAt != 0 && oldHdr.expireAt < now

     leastRecentUsed := int64(oldHdr.accessTime)*atomic.LoadInt64(&seg.totalCount) <= atomic.LoadInt64(&seg.totalTime)

     ifexpired || leastRecentUsed || consecutiveEvacuate > 5 {

     // 可以回收

        seg.delEntryPtrByOffset(oldHdr.slotId, oldHdr.hash16, oldOff)

        ifoldHdr.slotId == slotId {

           slotModified = true

        }

        consecutiveEvacuate = 0

        atomic.AddInt64(&seg.totalTime, -int64(oldHdr.accessTime))

        atomic.AddInt64(&seg.totalCount, -1)

        seg.vacuumLen += oldEntryLen

        ifexpired {

           atomic.AddInt64(&seg.totalExpired, 1)

        } else{

           atomic.AddInt64(&seg.totalEvacuate, 1)

        }

     } else{

        // evacuate an old entry that has been accessed recently for better cache hit rate.

        newOff := seg.rb.Evacuate(oldOff, int(oldEntryLen))

        seg.updateEntryPtr(oldHdr.slotId, oldHdr.hash16, oldOff, newOff)

        consecutiveEvacuate++

        atomic.AddInt64(&seg.totalEvacuate, 1)

     }

  }

}

freecache的Get流程相对来说简单点，通过hash找到对应segment，通过slotId找到对应索引slot，然后通过二分+遍历寻找数据，如果找不到直接返回ErrNotFound，否则更新一些time指标。Get流程还会更新缓存命中率相关指标。

func (cache *Cache) Get(key []byte) (value []byte, err error) {

  hashVal := hashFunc(key)

  segID := hashVal & segmentAndOpVal

  cache.locks[segID].Lock()

  value, _, err = cache.segments[segID].get(key, nil, hashVal, false)

  cache.locks[segID].Unlock()

  return

}

func (seg *segment) get(key, buf []byte, hashVal uint64, peek bool) (value []byte, expireAt uint32, err error) {

  hdr, ptr, err := seg.locate(key, hashVal, peek) // hash+定位查找

  iferr != nil {

     return

  }

  expireAt = hdr.expireAt

  ifcap(buf) >= int(hdr.valLen) {

     value = buf[:hdr.valLen]

  } else{

     value = make([]byte, hdr.valLen)

  }

  seg.rb.ReadAt(value, ptr.offset+ENTRY_HDR_SIZE+int64(hdr.keyLen))

}

定位到数据之后，读取ringbuf即可，注意一般来说读取到的value是新创建的内存空间，因此涉及到[]byte数据的复制操作。

3 总结

从常见的几个缓存框架压测性能来看，Set性能差异较大但还不是数量级别的差距，Get性能差异不大，因此对于绝大多数场景来说不用太关注Set/Get性能，重点应该看功能是否满足业务需求和gc影响，性能压测比较见：https://golang2.eddycjy.com/posts/ch5/04-performance/

缓存有一个特殊的场景，那就是将数据全部缓存在内存，涉及到更新时都是全量更新(替换)，该场景下使用freecache，如果size未设置好可能导致部分数据被淘汰，是不符合预期的，这个一定要注意。为了使用freecache避免该问题，需要将size设置"足够大"，但也要注意其内存空间占用。

关注公众号：拾黑（shiheibook）了解更多

友情链接：

下软件就上简单下载站：https://www.jdsec.com/
四季很好，只要有你，文娱排行榜：https://www.yaopaiming.com/
让资讯触达的更精准有趣：https://www.0xu.cn/