一文了解Golang中的快取庫freecache

這篇文章帶大家了解一下Golang緩存，深入淺出的介紹一下Golang中的快取庫freecache，希望對大家有幫助！

go開發快取場景一般使用map或快取框架，為了執行緒安全性會使用sync.Map或執行緒安全的快取框架。

快取場景中如果資料量大於百萬級別，需要特別考慮資料類型對於gc的影響（注意string類型底層是指標Len Cap，因此也算是指標類型），如果快取key和value都是非指針類型的話就無需多慮了。 【相關推薦：Go影片教學】

但實際應用程式場景中，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)

   for i := 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()
   for i := 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)
    for i := 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#流程：

_ = 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)

   var hdrBuf [ENTRY_HDR_SIZE]byte
   hdr := (*entryHdr)(unsafe.Pointer(&hdrBuf[0]))
   if match { // 有数据更新操作
      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))
      if hdr.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.
      for hdr.valCap < hdr.valLen {
         hdr.valCap *= 2
      }
      if hdr.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))
      if hdr.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)
   if slotModified {
      // the slot has been modified during evacuation, we need to looked up for the &#39;idx&#39; 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) {
   var oldHdrBuf [ENTRY_HDR_SIZE]byte
   consecutiveEvacuate := 0
   for seg.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)
      if oldHdr.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)
      if expired || leastRecentUsed || consecutiveEvacuate > 5 {
      // 可以回收
         seg.delEntryPtrByOffset(oldHdr.slotId, oldHdr.hash16, oldOff)
         if oldHdr.slotId == slotId {
            slotModified = true
         }
         consecutiveEvacuate = 0
         atomic.AddInt64(&seg.totalTime, -int64(oldHdr.accessTime))
         atomic.AddInt64(&seg.totalCount, -1)
         seg.vacuumLen += oldEntryLen
         if expired {
            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+定位查找
   if err != nil {
      return
   }
   expireAt = hdr.expireAt
   if cap(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設定"足夠大"，但也要注意其記憶體空間佔用。

更多程式相關知識，請造訪：程式設計教學！ ！

以上是一文了解Golang中的快取庫freecache的詳細內容。更多資訊請關注PHP中文網其他相關文章！