前言

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

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

現(xiàn)在這一版本算是一個1.0，后續(xù)會繼續(xù)進行優(yōu)化和迭代。

第一步：抽象接口

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

// 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)：使用該方法存儲的數(shù)據(jù)使用默認的過期時間，如果清除過期的異步任務(wù)沒有enable，那么就永不過期，否則默認過期時間為10min。
• Get(key string) ([]byte, error)：根據(jù)key獲取對象內(nèi)容，如果數(shù)據(jù)過期了會在這一步刪除。
• SetWithTime(key string, value []byte, expired time.Duration)：存儲對象是使用自定義過期時間
• Delete(key string) error：根據(jù)key刪除對應(yīng)的緩存數(shù)據(jù)
• Len() int：獲取緩存的對象數(shù)量
• Capacity() int：獲取當(dāng)前緩存的容量
• Close() error：關(guān)閉緩存
• Stats() Stats：緩存監(jiān)控數(shù)據(jù)
• GetKeyHit(key string) int64：獲取key的命中率數(shù)據(jù)

第二步：定義緩存對象

第一步我們抽象好了接口，下面就要定義一個緩存對象實例實現(xiàn)接口，先看定義結(jié)構(gòu)：

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：分片要用的哈希函數(shù)，用戶可以自行定義，實現(xiàn)HashFunc接口即可，默認使用fnv算法。
• bucketCount：分片的數(shù)量，一定要是偶數(shù)，默認分片數(shù)為256。
• bucketMask：因為分片數(shù)是偶數(shù)，所以可以分片時可以使用位運算代替取余提升性能效率，hashValue % bucketCount == hashValue & bucketCount - 1。
• segments：分片對象，每個分片的對象結(jié)構(gòu)我們在后面介紹。
• locks：每個分片的讀寫鎖
• close：關(guān)閉緩存對象時通知其他goroutine暫停

接下來我們來寫cache對象的構(gòu)造函數(shù)：

// 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編程模式，我們的構(gòu)造函數(shù)主要做三件事：

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

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

第三步：定義分片結(jié)構(gòu)

每個分片結(jié)構(gòu)如下：

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

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

接下來我們再來看一下每個分片的構(gòu)造函數(shù)：

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), 
 } 
} 

這里主要注意一點：

我們要根據(jù)每個片區(qū)的緩存數(shù)據(jù)大小來計算出容量，與上文的緩存對象初始化步驟對應(yīng)上了。

第四步：定義緩存結(jié)構(gòu)

緩存對象現(xiàn)在也構(gòu)造好了，接下來就是本地緩存的核心：定義緩存結(jié)構(gòu)。

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

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

每個緩存的封裝結(jié)構(gòu)如下：

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

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：緩存結(jié)構(gòu)的最大容量
• index：索引，記錄緩存所在的位置的索引
• count：記錄緩存數(shù)量
• availableSpace：記錄"蟲洞"，當(dāng)緩存對象被刪除時記錄下空閑位置的索引，方便后面容量滿了后使用"蟲洞"
• placeholder：記錄緩存對象的索引，迭代清除過期緩存可以用上。

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

第五步：完善向緩存寫入數(shù)據(jù)方法

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

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) 
 } 
} 

流程分析如下：

根據(jù)key計算哈希值，然后根據(jù)分片數(shù)獲取對應(yīng)分片位置

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

封裝存儲結(jié)構(gòu)，根據(jù)過期時間戳、key長度、哈希大小、緩存對象進行封裝

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

第六步：完善從緩存讀取數(shù)據(jù)方法

第一步根據(jù)key計算哈希值，再根據(jù)分片數(shù)獲取對應(yīng)的分片位置：

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 
} 

第二步執(zhí)行分片方法獲取緩存數(shù)據(jù)：

• 先根據(jù)哈希值判斷key是否存在于緩存中，不存返回key沒有找到
• 從緩存中讀取數(shù)據(jù)得到緩存中的key判斷是否發(fā)生哈希沖突
• 判斷緩存對象是否過期，過期刪除緩存數(shù)據(jù)(可以根據(jù)業(yè)務(wù)優(yōu)化需要是否返回當(dāng)前過期數(shù)據(jù))
• 在每個記錄緩存監(jiān)控數(shù)據(jù)

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)) 
} 

運行結(jié)果：

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

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

參考文章

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

總結(jié)

實現(xiàn)篇到這里就結(jié)束了，但是這個項目的編碼仍未結(jié)束，我會繼續(xù)以此版本為基礎(chǔ)不斷迭代優(yōu)化，該本地緩存的優(yōu)點：

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

待優(yōu)化點：

• 沒有使用高效的緩存淘汰算法，可能會導(dǎo)致熱點數(shù)據(jù)被頻繁刪除
• 定時刪除過期數(shù)據(jù)會導(dǎo)致鎖持有時間過長，需要優(yōu)化
• 關(guān)閉緩存實例需要優(yōu)化處理方式
• 根據(jù)業(yè)務(wù)場景進行優(yōu)化(特定業(yè)務(wù)場景)

迭代點：

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

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

好啦，本文到這里就結(jié)束了，我是asong，我們下期見。