Golang中的缓存库freecache怎么用(freecache,golang,编程语言)

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为例分析下其实现原理，代码示例如下：

funcmain(){cacheSize:=100*1024*1024cache:=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,GCtook:%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,Gettook:%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（注意切片为指针类型）。

typesegmentstruct{rbRingBuf//ringbufferthatstoresdatasegIdint_uint32//占位missCountint64hitCountint64entryCountint64totalCountint64//numberofentriesinringbuffer,includingdeletedentries.totalTimeint64//usedtocalculateleastrecentusedentry.timerTimer//TimergivingcurrenttimetotalEvacuateint64//usedfordebugtotalExpiredint64//usedfordebugoverwritesint64//usedfordebugtouchedint64//usedfordebugvacuumLenint64//uptovacuumLen,newdatacanbewrittenwithoutoverwritingolddata.slotLens[256]int32//Theactuallengthforeveryslot.slotCapint32//maxnumberofentrypointersaslotcanhold.slotsData[]entryPtr//索引指针}funcNewCacheCustomTimer(sizeint,timerTimer)(cache*Cache){cache=new(Cache)fori:=0;i<segmentCount;i++{cache.segments[i]=newSegment(size/segmentCount,i,timer)}}funcnewSegment(bufSizeint,segIdint,timerTimer)(segsegment){seg.rb=NewRingBuf(bufSize,0)seg.segId=segIdseg.timer=timerseg.vacuumLen=int64(bufSize)seg.slotCap=1seg.slotsData=make([]entryPtr,256*seg.slotCap)//每个slotData初始化256个单位大小}

2 读写流程

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

_=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,expireSecondsint)(errerror){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,hashValuint64,expireSecondsint)(errerror){slotId:=uint8(hashVal>>8)hash26:=uint16(hashVal>>16)slot:=seg.getSlot(slotId)idx,match:=seg.lookup(slot,hash26,key)varhdrBuf[ENTRY_HDR_SIZE]bytehdr:=(*entryHdr)(unsafe.Pointer(&hdrBuf[0]))ifmatch{//有数据更新操作matchedPtr:=&slot[idx]seg.rb.ReadAt(hdrBuf[:],matchedPtr.offset)hdr.slotId=slotIdhdr.hash26=hash26hdr.keyLen=uint16(len(key))originAccessTime:=hdr.accessTimehdr.accessTime=nowhdr.expireAt=expireAthdr.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//increasecapacityandlimitentrylen.forhdr.valCap<hdr.valLen{hdr.valCap*=2}ifhdr.valCap>uint32(maxKeyValLen-len(key)){hdr.valCap=uint32(maxKeyValLen-len(key))}}else{//无数据hdr.slotId=slotIdhdr.hash26=hash26hdr.keyLen=uint16(len(key))hdr.accessTime=nowhdr.expireAt=expireAthdr.valLen=uint32(len(value))hdr.valCap=uint32(len(value))ifhdr.valCap==0{//avoidinfiniteloopwhenincreasingcapacity.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{//theslothasbeenmodifiedduringevacuation,weneedtolookedupforthe'idx'again.//otherwisetherewouldbeindexoutofbounderror.slot=seg.getSlot(slotId)idx,match=seg.lookup(slot,hash26,key)//assert(match==false)}newOff:=seg.rb.End()seg.insertEntryPtr(slotId,hash26,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-=entryLenreturn}

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(entryLenint64,slotIduint8,nowuint32)(slotModifiedbool){varoldHdrBuf[ENTRY_HDR_SIZE]byteconsecutiveEvacuate:=0forseg.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=0atomic.AddInt64(&seg.totalTime,-int64(oldHdr.accessTime))atomic.AddInt64(&seg.totalCount,-1)seg.vacuumLen+=oldEntryLencontinue}expired:=oldHdr.expireAt!=0&&oldHdr.expireAt<nowleastRecentUsed:=int64(oldHdr.accessTime)*atomic.LoadInt64(&seg.totalCount)<=atomic.LoadInt64(&seg.totalTime)ifexpired||leastRecentUsed||consecutiveEvacuate>5{//可以回收seg.delEntryPtrByOffset(oldHdr.slotId,oldHdr.hash26,oldOff)ifoldHdr.slotId==slotId{slotModified=true}consecutiveEvacuate=0atomic.AddInt64(&seg.totalTime,-int64(oldHdr.accessTime))atomic.AddInt64(&seg.totalCount,-1)seg.vacuumLen+=oldEntryLenifexpired{atomic.AddInt64(&seg.totalExpired,1)}else{atomic.AddInt64(&seg.totalEvacuate,1)}}else{//evacuateanoldentrythathasbeenaccessedrecentlyforbettercachehitrate.newOff:=seg.rb.Evacuate(oldOff,int(oldEntryLen))seg.updateEntryPtr(oldHdr.slotId,oldHdr.hash26,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,errerror){hashVal:=hashFunc(key)segID:=hashVal&segmentAndOpValcache.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,hashValuint64,peekbool)(value[]byte,expireAtuint32,errerror){hdr,ptr,err:=seg.locate(key,hashVal,peek)//hash+定位查找iferr!=nil{return}expireAt=hdr.expireAtifcap(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数据的复制操作。