動手實現一個Localcache-實現篇

作者： AsongGo 2022-01-04 11:08:02

開發前端

使用二維數組存儲數據的相比于bigcache的優勢在于可以直接根據索引刪除對應的數據，雖然也會有蟲洞的問題，但是我們可以記錄下來蟲洞的索引，不斷填充。

前言

哈嘍，大家好，我是asong，經過了前面兩篇的介紹，我們已經基本了解該如何設計一個本地緩存了，本文就是這個系列的終結篇，自己動手實現一個本地緩存，接下來且聽我細細道來!!!

本文代碼已經上傳到github：https://github.com/asong2020/go-localcache

現在這一版本算是一個1.0，后續會繼續進行優化和迭代。

第一步：抽象接口

第一步很重要，以面向接口編程為原則，我們先抽象出來要暴露給用戶的方法，給用戶提供簡單易懂的方法，因此我抽象出來的結果如下：

// ICache abstract interface 
type ICache interface { 
 // Set value use default expire time. default does not expire. 
 Set(key string, value []byte) error 
 // Get value if find it. if value already expire will delete. 
 Get(key string) ([]byte, error) 
 // SetWithTime set value with expire time 
 SetWithTime(key string, value []byte, expired time.Duration) error 
 // Delete manual removes the key 
 Delete(key string) error 
 // Len computes number of entries in cache 
 Len() int 
 // Capacity returns amount of bytes store in the cache. 
 Capacity() int 
 // Close is used to signal a shutdown of the cache when you are done with it. 
 // This allows the cleaning goroutines to exit and ensures references are not 
 // kept to the cache preventing GC of the entire cache. 
 Close() error 
 // Stats returns cache's statistics 
 Stats() Stats 
 // GetKeyHit returns key hit 
 GetKeyHit(key string) int64 
}

Set(key string, value []byte)：使用該方法存儲的數據使用默認的過期時間，如果清除過期的異步任務沒有enable，那么就永不過期，否則默認過期時間為10min。
Get(key string) ([]byte, error)：根據key獲取對象內容，如果數據過期了會在這一步刪除。
SetWithTime(key string, value []byte, expired time.Duration)：存儲對象是使用自定義過期時間
Delete(key string) error：根據key刪除對應的緩存數據
Len() int：獲取緩存的對象數量
Capacity() int：獲取當前緩存的容量
Close() error：關閉緩存
Stats() Stats：緩存監控數據
GetKeyHit(key string) int64：獲取key的命中率數據

第二步：定義緩存對象

第一步我們抽象好了接口，下面就要定義一個緩存對象實例實現接口，先看定義結構：

type cache struct { 
 // hashFunc represents used hash func 
 hashFunc HashFunc 
 // bucketCount represents the number of segments within a cache instance. value must be a power of two. 
 bucketCount uint64 
 // bucketMask is bitwise AND applied to the hashVal to find the segment id. 
 bucketMask uint64 
 // segment is shard 
 segments []*segment 
 // segment lock 
 locks    []sync.RWMutex 
 // close cache 
 close chan struct{} 
}

hashFunc：分片要用的哈希函數，用戶可以自行定義，實現HashFunc接口即可，默認使用fnv算法。
bucketCount：分片的數量，一定要是偶數，默認分片數為256。
bucketMask：因為分片數是偶數，所以可以分片時可以使用位運算代替取余提升性能效率，hashValue % bucketCount == hashValue & bucketCount - 1。
segments：分片對象，每個分片的對象結構我們在后面介紹。
locks：每個分片的讀寫鎖
close：關閉緩存對象時通知其他goroutine暫停

接下來我們來寫cache對象的構造函數：

// NewCache constructor cache instance 
func NewCache(opts ...Opt) (ICache, error) { 
 options := &options{ 
  hashFunc: NewDefaultHashFunc(), 
  bucketCount: defaultBucketCount, 
  maxBytes: defaultMaxBytes, 
  cleanTime: defaultCleanTIme, 
  statsEnabled: defaultStatsEnabled, 
  cleanupEnabled: defaultCleanupEnabled, 
 } 
 for _, each := range opts{ 
  each(options) 
 } 
 
 if !isPowerOfTwo(options.bucketCount){ 
  return nil, errShardCount 
 } 
 
  if options.maxBytes <= 0 { 
  return nil, ErrBytes 
 } 
   
 segments := make([]*segment, options.bucketCount) 
 locks := make([]sync.RWMutex, options.bucketCount) 
 
 maxSegmentBytes := (options.maxBytes + options.bucketCount - 1) / options.bucketCount 
 for index := range segments{ 
  segments[index] = newSegment(maxSegmentBytes, options.statsEnabled) 
 } 
 
 c := &cache{ 
  hashFunc: options.hashFunc, 
  bucketCount: options.bucketCount, 
  bucketMask: options.bucketCount - 1, 
  segments: segments, 
  locks: locks, 
  close: make(chan struct{}), 
 } 
    if options.cleanupEnabled { 
  go c.cleanup(options.cleanTime) 
 } 
  
 return c, nil 
}

這里為了更好的擴展，我們使用Options編程模式，我們的構造函數主要做三件事：

前置參數檢查，對于外部傳入的參數，我們還是要做基本的校驗
分片對象初始化
構造緩存對象

這里構造緩存對象時我們要先計算每個分片的容量，默認整個本地緩存256M的數據，然后在平均分到每一片區內，用戶可以自行選擇要緩存的數據大小。

第三步：定義分片結構

每個分片結構如下：

type segment struct { 
 hashmap map[uint64]uint32 
 entries buffer.IBuffer 
 clock   clock 
 evictList  *list.List 
 stats IStats 
}

hashmp：存儲key所對應的存儲索引
entries：存儲key/value的底層結構，我們在第四步的時候介紹，也是代碼的核心部分。
clock：定義時間方法
evicList：這里我們使用一個隊列來記錄old索引，當容量不足時進行刪除(臨時解決方案，當前存儲結構不適合使用LRU淘汰算法)
stats：緩存的監控數據。

接下來我們再來看一下每個分片的構造函數：

func newSegment(bytes uint64, statsEnabled bool) *segment { 
 if bytes == 0 { 
  panic(fmt.Errorf("bytes cannot be zero")) 
 } 
 if bytes >= maxSegmentSize{ 
  panic(fmt.Errorf("too big bytes=%d; should be smaller than %d", bytes, maxSegmentSize)) 
 } 
 capacity := (bytes + segmentSize - 1) / segmentSize 
 entries := buffer.NewBuffer(int(capacity)) 
 entries.Reset() 
 return &segment{ 
  entries: entries, 
  hashmap: make(map[uint64]uint32), 
  clock:   &systemClock{}, 
  evictList: list.New(), 
  stats: newStats(statsEnabled), 
 } 
}

這里主要注意一點：

我們要根據每個片區的緩存數據大小來計算出容量，與上文的緩存對象初始化步驟對應上了。

第四步：定義緩存結構

緩存對象現在也構造好了，接下來就是本地緩存的核心：定義緩存結構。

bigcache、fastcache、freecache都使用字節數組代替map存儲緩存數據，從而減少GC壓力，所以我們也可以借鑒其思想繼續保持使用字節數組，這里我們使用二維字節切片存儲緩存數據key/value;畫個圖表示一下：

每個緩存的封裝結構如下：

基本思想已經明確，接下來看一下我們對存儲層的封裝：

type Buffer struct { 
 array [][]byte 
 capacity int 
 index int 
 // maxCount = capacity - 1 
 count int 
 // availableSpace If any objects are removed after the buffer is full, the idle index is logged. 
 // Avoid array "wormhole" 
 availableSpace map[int]struct{} 
 // placeholder record the index that buffer has stored. 
 placeholder map[int]struct{} 
}

array [][]byte：存儲緩存對象的二維切片
capacity：緩存結構的最大容量
index：索引，記錄緩存所在的位置的索引
count：記錄緩存數量
availableSpace：記錄"蟲洞"，當緩存對象被刪除時記錄下空閑位置的索引，方便后面容量滿了后使用"蟲洞"
placeholder：記錄緩存對象的索引，迭代清除過期緩存可以用上。

向buffer寫入數據的流程(不貼代碼了)：

第五步：完善向緩存寫入數據方法

上面我們定義好了所有需要的結構，接下來就是填充我們的寫入緩存方法就可以了：

func (c *cache) Set(key string, value []byte) error  { 
 hashKey := c.hashFunc.Sum64(key) 
 bucketIndex := hashKey&c.bucketMask 
 c.locks[bucketIndex].Lock() 
 defer c.locks[bucketIndex].Unlock() 
 err := c.segments[bucketIndex].set(key, hashKey, value, defaultExpireTime) 
 return err 
} 
 
func (s *segment) set(key string, hashKey uint64, value []byte, expireTime time.Duration) error { 
 if expireTime <= 0{ 
  return ErrExpireTimeInvalid 
 } 
 expireAt := uint64(s.clock.Epoch(expireTime)) 
 
 if previousIndex, ok := s.hashmap[hashKey]; ok { 
  if err := s.entries.Remove(int(previousIndex)); err != nil{ 
   return err 
  } 
  delete(s.hashmap, hashKey) 
 } 
 
 entry := wrapEntry(expireAt, key, hashKey, value) 
 for { 
  index, err := s.entries.Push(entry) 
  if err == nil { 
   s.hashmap[hashKey] = uint32(index) 
   s.evictList.PushFront(index) 
   return nil 
  } 
  ele := s.evictList.Back() 
  if err := s.entries.Remove(ele.Value.(int)); err != nil{ 
   return err 
  } 
  s.evictList.Remove(ele) 
 } 
}

流程分析如下：

根據key計算哈希值，然后根據分片數獲取對應分片位置

如果當前緩存中存在相同的key，則先刪除，在重新插入，會刷新過期時間

封裝存儲結構，根據過期時間戳、key長度、哈希大小、緩存對象進行封裝

將數據存入緩存，如果緩存失敗，移除最老的數據后再次重試

第六步：完善從緩存讀取數據方法

第一步根據key計算哈希值，再根據分片數獲取對應的分片位置：

func (c *cache) Get(key string) ([]byte, error)  { 
 hashKey := c.hashFunc.Sum64(key) 
 bucketIndex := hashKey&c.bucketMask 
 c.locks[bucketIndex].RLock() 
 defer c.locks[hashKey&c.bucketMask].RUnlock() 
 entry, err := c.segments[bucketIndex].get(key, hashKey) 
 if err != nil{ 
  return nil, err 
 } 
 return entry,nil 
}

第二步執行分片方法獲取緩存數據：

先根據哈希值判斷key是否存在于緩存中，不存返回key沒有找到
從緩存中讀取數據得到緩存中的key判斷是否發生哈希沖突
判斷緩存對象是否過期，過期刪除緩存數據(可以根據業務優化需要是否返回當前過期數據)
在每個記錄緩存監控數據

func (s *segment) getWarpEntry(key string, hashKey uint64) ([]byte,error) { 
 index, ok := s.hashmap[hashKey] 
 if !ok { 
  s.stats.miss() 
  return nil, ErrEntryNotFound 
 } 
 entry, err := s.entries.Get(int(index)) 
 if err != nil{ 
  s.stats.miss() 
  return nil, err 
 } 
 if entry == nil{ 
  s.stats.miss() 
  return nil, ErrEntryNotFound 
 } 
 
 if entryKey := readKeyFromEntry(entry); key != entryKey { 
  s.stats.collision() 
  return nil, ErrEntryNotFound 
 } 
 return entry, nil 
} 
 
func (s *segment) get(key string, hashKey uint64) ([]byte, error) { 
 currentTimestamp := s.clock.TimeStamp() 
 entry, err := s.getWarpEntry(key, hashKey) 
 if err != nil{ 
  return nil, err 
 } 
 res := readEntry(entry) 
 
 expireAt := int64(readExpireAtFromEntry(entry)) 
 if currentTimestamp - expireAt >= 0{ 
  _ = s.entries.Remove(int(s.hashmap[hashKey])) 
  delete(s.hashmap, hashKey) 
  return nil, ErrEntryNotFound 
 } 
 s.stats.hit(key) 
 
 return res, nil 
}

第七步：來個測試用例體驗一下

先來個簡單的測試用例測試一下：

func (h *cacheTestSuite) TestSetAndGet() { 
 cache, err := NewCache() 
 assert.Equal(h.T(), nil, err) 
 key := "asong" 
 value := []byte("公眾號：Golang夢工廠") 
 
 err = cache.Set(key, value) 
 assert.Equal(h.T(), nil, err) 
 
 res, err := cache.Get(key) 
 assert.Equal(h.T(), nil, err) 
 assert.Equal(h.T(), value, res) 
 h.T().Logf("get value is %s", string(res)) 
}

運行結果：

=== RUN   TestCacheTestSuite 
=== RUN   TestCacheTestSuite/TestSetAndGet 
    cache_test.go:33: get value is 公眾號：Golang夢工廠 
--- PASS: TestCacheTestSuite (0.00s) 
    --- PASS: TestCacheTestSuite/TestSetAndGet (0.00s) 
PASS

大功告成，基本功能通了，剩下就是跑基準測試、優化、迭代了(不在文章贅述了，可以關注github倉庫最新動態)。

參考文章

https://github.com/allegro/bigcache
https://github.com/VictoriaMetrics/fastcache
https://github.com/coocood/freecache
https://github.com/patrickmn/go-cache

總結

實現篇到這里就結束了，但是這個項目的編碼仍未結束，我會繼續以此版本為基礎不斷迭代優化，該本地緩存的優點：

實現簡單、提供給用戶的方法簡單易懂
使用二維切片作為存儲結構，避免了不能刪除底層數據的缺點，也在一定程度上避免了"蟲洞"問題。
測試用例齊全，適合作為小白的入門項目

待優化點：

沒有使用高效的緩存淘汰算法，可能會導致熱點數據被頻繁刪除
定時刪除過期數據會導致鎖持有時間過長，需要優化
關閉緩存實例需要優化處理方式
根據業務場景進行優化(特定業務場景)

迭代點：

添加異步加載緩存功能
...... (思考中)