bigcache

1.简介

1.1.bigcache

• bigcache是golang编写的高性能缓存库，通过数据分片缓解高并发下锁竞争问题，数据保存在ringbuffer规避频繁的GC回收。bigcache内部使用分片存储数据，每个分片使用hashmap存储key的索引，真正的数据通过编码后放在ringbuffer里。bigcache没有使用主流的lru或lfu缓存淘汰算法，而是使用覆盖写清理老数据。ringbuffer已满时，会删除老数据尝试写入新数据，通过GC垃圾回收清理过期数据。当然，bigcache数据存储的是[]byte类型，导致业务使用场景受限，频繁的序列化和反序列化在一定量级下会加重CPU负担。

• 

• 补充

  1.ringbuffer可以使用有名或匿名的mmap(堆外内存)实现，但mmap和[]byte的GC开销区别不大

  2.文件mmap映射实现ringbuffer时，系统的频繁文件读写势必涉及page cache淘汰，造成mmap构建的ringbuffer受影响

1.2.示例

• func main() {
      config := bigcache.Config {
              // 预设多少个数据分片，其大小必须是 2 的幂次方，因为这里使用位运算取摸，而非使用 %.
              Shards: 1024,
  
              // 缓存对象的生命周期，也就是过期时长
              LifeWindow: 10 * time.Minute,
  
              // 垃圾回收的运行周期，每隔 5 分钟尝试进行一次垃圾回收.
              CleanWindow: 5 * time.Minute,
  
              // rps * lifeWindow, used only in initial memory allocation
              MaxEntriesInWindow: 1000 * 10 * 60,
  
              // 设定的 value 的大小
              MaxEntrySize: 500,
  
              // bigcache 缓存的大小，单位是 MB.
              // 注意这是总大小，每个分片的大小则需要除以分片数，当为 0 时不限制。
              HardMaxCacheSize: 8192,
      }
  
      // 构建 bigcache 缓存对象
      cache, initErr := bigcache.New(context.Background(), config)
      if initErr != nil {
          log.Fatal(initErr)
      }
  
      // 写数据
      cache.Set("my-unique-key", []byte("value"))
  
      // 读数据
      if entry, err := cache.Get("my-unique-key"); err == nil {
          fmt.Println(string(entry))
      }
  }
  

2.原理分析

2.1.结构

• bigcache的数据读写涉及到ringbuffer的定位及存取，数据会存在ringbuffer中，hashmap中记录key和ringbuffer索引的关系，读取时会从hashmap里获取key的ringbuffer索引，然后从ringbuffer获取数据。

• type BigCache struct {
      // 分片
  	shards     []*cacheShard
      // 缓存对象过期时间,ttl
  	lifeWindow uint64
      // 时钟对象
  	clock      clock
      // 使用的hash算法
  	hash       Hasher
  	config     Config
      // 取模掩码
  	shardMask  uint64
  	close      chan struct{}
  }
  

• 

• type cacheShard struct {
      // 键key的hashcode，值为ringbuffer的offset
  	hashmap     map[uint64]uint32
      // 使用ringbuffer构建的队列
  	entries     queue.BytesQueue
  	lock        sync.RWMutex
      // 对象复用
  	entryBuffer []byte
      // 删除回调
  	onRemove    onRemoveCallback
  	// 缓存对象过期时间,ttl
  	lifeWindow   uint64
  }
  

• 

• 优化

  1.一般的取模采用%实现，但从汇编角度来看，&的汇编为3mov+1and+1sub，%的汇编为2mov+1cdp+1idiv，两者动作看起来区别不打。根据intel asm文档提到，&只需要5个CPU周期，%至少需要20个CPU周期。

  2.底层数据保存每个entry长度时，采用无符号varint编码减少空间占用。

  数字大小	uvarint编码需要的字节数
  <=127	1
  <=16383	2
  <=2097151	3
  <=268435455	4

2.2.Set操作

• set用来把数据添加到shard的ringbuffer里，同时通过hashmap记录[hash(key),index]索引，其内部将分片数据的添加委托给了shard的set实现。

• func (c *BigCache) Set(key string, entry []byte) error {
      // 通过fnv hash算法计算key的hashcode
  	hashedKey := c.hash.Sum64(key)
      // 通过位运算获取shard
  	shard := c.getShard(hashedKey)
      // 向分片添加数据
  	return shard.set(key, hashedKey, entry)
  }
  

• set数据时，会尝试删除旧的entry，然后尝试写入数据，写入失败后涉及到扩容或删除最老数据，这也说明了最老数据的清理是惰性的。

• func (s *cacheShard) set(key string, hashedKey uint64, entry []byte) error {
      // 获取当前时间戳
  	currentTimestamp := uint64(s.clock.Epoch())
  
      // 加锁
  	s.lock.Lock()
  
      // 从hashmap获取hash(key)的index
  	if previousIndex := s.hashmap[hashedKey]; previousIndex != 0 {
          // ringbuffer中根据index找到entry
  		if previousEntry, err := s.entries.Get(int(previousIndex)); err == nil {
              // 将之前entry的hashcode设为0,相当于标记删除
  			resetKeyFromEntry(previousEntry)
              // 从hashmap删除旧的hash(key)
  			delete(s.hashmap, hashedKey)
  		}
  	}
  
  	// 如果没有开启后台定时清理
  	if !s.cleanEnabled {
  		// 检查最老的一个entry是否过期, 如果过期就删除
  		if oldestEntry, err := s.entries.Peek(); err == nil {
  			s.onEvict(oldestEntry, currentTimestamp, s.removeOldestEntry)
  		}
  	}
  
      // 编码待写入ringbuffer里的结构
  	w := wrapEntry(currentTimestamp, hashedKey, key, entry, &s.entryBuffer)
  
  	for {
          // 编码后的数据写入ringbuffer
  		if index, err := s.entries.Push(w); err == nil {
              // 成功写入则更新hashmap索引映射
  			s.hashmap[hashedKey] = uint32(index)
  			s.lock.Unlock()
  			return nil
  		}
          // 移除最老的entry(弹出head位置数据,弹出后head后移)
  		if s.removeOldestEntry(NoSpace) != nil {
  			s.lock.Unlock()
  			return fmt.Errorf("entry is bigger than max shard size")
  		}
  	}
  }
  

• resetKeyFromEntry会把entry中hashcode置为0，entry[8:16]字节存储了数据的key hashcode，通过resetKeyFromEntry方法可以把hashcode置为0。

• func resetKeyFromEntry(data []byte) {
      binary.LittleEndian.PutUint64(data[timestampSizeInBytes:], 0)
  }
  

• GC垃圾回收通过时间戳判断数据是否过期或hashcode是否为0，通过iterator遍历缓存数据时，会过滤掉hashcode为0的数据，当删除数据时，会重置hashcode为0，通过key length拿到key。编码数据时，会根据前半部分的length写入后边的数据，因此获取value也可以通过偏移实现。

• const (
      timestampSizeInBytes = 8
      hashSizeInBytes      = 8
      keySizeInBytes       = 2
      headersSizeInBytes   = timestampSizeInBytes + hashSizeInBytes + keySizeInBytes
  )
  
  func wrapEntry(timestamp uint64, hash uint64, key string, entry []byte, buffer *[]byte) []byte {
  	keyLength := len(key)
  	blobLength := len(entry) + headersSizeInBytes + keyLength
  
      // 当前entry包装长度超出,重新初始化更长的buffer
  	if blobLength > len(*buffer) {
  		*buffer = make([]byte, blobLength)
  	}
  	blob := *buffer
  
      // 写入时间戳
  	binary.LittleEndian.PutUint64(blob, timestamp)
      // 时间戳后写入hashcode
  	binary.LittleEndian.PutUint64(blob[timestampSizeInBytes:], hash)
      // 紧跟写入key的长度
  	binary.LittleEndian.PutUint16(blob[timestampSizeInBytes+hashSizeInBytes:], uint16(keyLength))
      // 写入key
  	copy(blob[headersSizeInBytes:], key)
      // key之后紧跟写入entry
  	copy(blob[headersSizeInBytes+keyLength:], entry)
  
  	return blob[:blobLength]
  }
  

• ringbuffer中内存碎片过多或无法写入新entry时会触发扩容，扩容时按照2倍原则重新申请ringbuffer内存。

• func (q *BytesQueue) Push(data []byte) (int, error) {
      // 计算出entry需要长度
  	neededSize := getNeededSize(len(data))
  
      // tail后边没有足够空间
  	if !q.canInsertAfterTail(neededSize) {
          // 回绕,尝试在head前写
  		if q.canInsertBeforeHead(neededSize) {
              // 更新tail为1,代表底层数组循环了一圈
  			q.tail = leftMarginIndex
  		} else if q.capacity+neededSize >= q.maxCapacity && q.maxCapacity > 0 {
              // 超出最大限制
  			return -1, &queueError{"Full queue. Maximum size limit reached."}
  		} else {
              // 扩容
  			q.allocateAdditionalMemory(neededSize)
  		}
  	}
  
  	index := q.tail
  
      // 写入数据
  	q.push(data, neededSize)
  
  	return index, nil
  }
  

• canInsertAfterTail用于检查tail后是否存在可用区域存储entry，canInsertBeforeHead检查head前是否存在可用区域存储entry，其会判断两种情况：有效数据区连续、有效数据区分散。

• func (q *BytesQueue) canInsertAfterTail(need int) bool {
      // 已经满了
  	if q.full {
  		return false
  	}
      // 连续情况,判断tail到尾部是否还有空间
  	if q.tail >= q.head {
  		return q.capacity-q.tail >= need
  	}
  	// 分散情况,刚好装下need、扣减need后够装minimumHeaderSize
  	return q.head-q.tail == need || q.head-q.tail >= need+minimumHeaderSize
  }
  
  func (q *BytesQueue) canInsertBeforeHead(need int) bool {
      // 已经满了
  	if q.full {
  		return false
  	}
      // 连续情况,判断开始~head间是否还有空间(过期的回收或前边数据被删除)
  	if q.tail >= q.head {
  		return q.head-leftMarginIndex == need || q.head-leftMarginIndex >= need+minimumHeaderSize
  	}
      // 分散情况,刚好装下need、扣减need后够装minimumHeaderSize
  	return q.head-q.tail == need || q.head-q.tail >= need+minimumHeaderSize
  }
  

• 

• tail后或head前没有空间存储entry时，调用allocateAdditionalMemory扩容，将旧的entry复制到容量倍增的新区域。

• func (q *BytesQueue) allocateAdditionalMemory(minimum int) {
  	start := time.Now()
      // 容量小于最小阈值时
  	if q.capacity < minimum {
          // 容量加上最小阈值再扩容,避免之后的频繁扩容
  		q.capacity += minimum
  	}
      // 容量二倍扩容
  	q.capacity = q.capacity * 2
  	if q.capacity > q.maxCapacity && q.maxCapacity > 0 {
  		// 超出最大容量则重置为最大容量
          q.capacity = q.maxCapacity
  	}
  
  	oldArray := q.array
      // 重新申请ringbuffer内存
  	q.array = make([]byte, q.capacity)
  	
      // 旧环存在数据
  	if leftMarginIndex != q.rightMargin {
          // copy旧环数据(这样拷贝是为了hashmap中的索引映射不失效)
  		copy(q.array, oldArray[:q.rightMargin])
  		if q.tail <= q.head {
              // 数据回绕,但是tail到head间存不下其他entry,为了避免数据不连续,将该空区域填一个空字节切片
  			if q.tail != q.head {
  				q.push(make([]byte, q.head-q.tail), q.head-q.tail)
  			}
  			// 重置首尾指针位置
  			q.head = leftMarginIndex
  			q.tail = q.rightMargin
  		}
  	}
  
  	q.full = false
  
  	if q.verbose {
  		log.Printf("Allocated new queue in %s; Capacity: %d \n", time.Since(start), q.capacity)
  	}
  }
  

• 空间够用或扩容后，就可以将包装好的entry写入ringbuffer，此时会将tail后移， 更新右边界。

• func (q *BytesQueue) push(data []byte, len int) {
  	headerEntrySize := binary.PutUvarint(q.headerBuffer, uint64(len))
  	// entry占用的字节长度
      q.copy(q.headerBuffer, headerEntrySize)
  	// 写入数据
  	q.copy(data, len-headerEntrySize)
  
      // tail未越界,更新右边界
  	if q.tail > q.head {
  		q.rightMargin = q.tail
  	}
      // 首尾指针重叠,说明ringbuffer满了
  	if q.tail == q.head {
  		q.full = true
  	}
  	// 更新环中entry数量
  	q.count++
  }
  

• 注意

  1.copy(q.array, oldArray[:q.rightMargin])会将右边界前的数据都复制到新环，末端存在空区域时会忽略

  2.q.push(make([]byte, q.head-q.tail), q.head-q.tail)在不足以写entry时追加一块空字节切片，保证数据连续

  3.q.copy(data, needsize)会在写入元素后更新tail位置及右边界rightMargin位置直至回绕

2.3.Get操作

• 获取数据时会根据hash获取shard，然后在hashmap中拿到key在ringbuffer索引，最后定位并读取entry。

• func (c *BigCache) Get(key string) ([]byte, error) {
      // 计算key的哈希值
  	hashedKey := c.hash.Sum64(key)
      // 获取对应分片
  	shard := c.getShard(hashedKey)
      // 从分片的ringbuffer中获取数据
  	return shard.get(key, hashedKey)
  }
  

• get用来从shard获取数据，其先从ringbuffer里获取编码过的数据，然后通过解码获得value。

• func (s *cacheShard) get(key string, hashedKey uint64) ([]byte, error) {
  	s.lock.RLock()
      // 根据hashmap中保存的key在ringbuffer索引拿到entry,然后从ringbuffer读取数据
  	wrappedEntry, err := s.getWrappedEntry(hashedKey)
  	if err != nil {
  		s.lock.RUnlock()
  		return nil, err
  	}
      // 从entry解码的key与当前key不一致说明hash冲突,数据不存在
  	if entryKey := readKeyFromEntry(wrappedEntry); key != entryKey {
  		s.lock.RUnlock()
  		s.collision()
  		}
  		return nil, ErrEntryNotFound
  	}
  	// 获取entry
  	entry := readEntry(wrappedEntry)
  	s.lock.RUnlock()
  	s.hit(hashedKey)
  
  	return entry, nil
  }
  

• getWrappedEntry从ringbuffer获取数据wrappedEntry，其实就是未解码的entry。

• func (s *cacheShard) getWrappedEntry(hashedKey uint64) ([]byte, error) {
      // 获取ringbuffer索引
  	itemIndex := s.hashmap[hashedKey]
  
  	if itemIndex == 0 {
  		s.miss()
  		return nil, ErrEntryNotFound
  	}
      // 从index开始读完整个entry(其实就是Set时binary.PutUvarint存了entry长度)
  	wrappedEntry, err := s.entries.Get(int(itemIndex))
  	if err != nil {
  		s.miss()
  		return nil, err
  	}
  
  	return wrappedEntry, err
  }
  

• readKeyFromEntry从entry中读取key，读取时会跳过时间戳、hashcode、key length，然后读取key length个字节就拿到key了。

• func readKeyFromEntry(data []byte) string {
      // 获取uint16编码的key长度
  	length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
  	dst := make([]byte, length)
      // 跳过时间戳、hashcode、key长度读取后边的key
  	copy(dst, data[headersSizeInBytes:headersSizeInBytes+length])
  	return bytesToString(dst)
  }
  

• readEntry用于读取value，其实就是跳过前边字节长度后读value，时间戳、hashcode、key length、key、value都会按照一定格式编码存储，都有各自长度。

• func readEntry(data []byte) []byte {
      // 获取key的长度
  	length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
  	dst := make([]byte, len(data)-int(headersSizeInBytes+length))
      // 读取value
  	copy(dst, data[headersSizeInBytes+length:])
  
  	return dst
  }
  

2.4.Delete操作

• Delete操作非常轻量，会把entry置为标记删除，删除索引，entry的真正删除会在过期GC、容量不足从head弹出进行。

• func (c *BigCache) Delete(key string) error {
      // 计算key的hash
  	hashedKey := c.hash.Sum64(key)
      // 获取分片
  	shard := c.getShard(hashedKey)
      // ringbuffer中的entry标记删除
  	return shard.del(hashedKey)
  }
  

• del删除分片数据时，会先加读锁检查entry是否存在，存在时拿到entry位置，将entry的hashcode置为0，然后从hashmap删除key关联索引。

• func (s *cacheShard) del(hashedKey uint64) error {
  	s.lock.RLock()
  	{
          // 获取ringbuffer中的key索引
  		itemIndex := s.hashmap[hashedKey]
  
  		if itemIndex == 0 {
  			s.lock.RUnlock()
  			s.delmiss()
  			return ErrEntryNotFound
  		}
  
  		if err := s.entries.CheckGet(int(itemIndex)); err != nil {
  			s.lock.RUnlock()
  			s.delmiss()
  			return err
  		}
  	}
  	s.lock.RUnlock()
  
  	s.lock.Lock()
  	{
  		// 再次获取索引回检
  		itemIndex := s.hashmap[hashedKey]
  
  		if itemIndex == 0 {
  			s.lock.Unlock()
  			s.delmiss()
  			return ErrEntryNotFound
  		}
  
          // 获取entry
  		wrappedEntry, err := s.entries.Get(int(itemIndex))
  		if err != nil {
  			s.lock.Unlock()
  			s.delmiss()
  			return err
  		}
  
          // 删除索引
  		delete(s.hashmap, hashedKey)
          // 回调
  		s.onRemove(wrappedEntry, Deleted)
  		if s.statsEnabled {
  			delete(s.hashmapStats, hashedKey)
  		}
          // entry的hashcode置为0
  		resetKeyFromEntry(wrappedEntry)
  	}
  	s.lock.Unlock()
  
      // 删除命中统计
  	s.delhit()
  	return nil
  }
  

2.5.GC设计

• bigcache的GC垃圾回收是指淘汰清理过期的数据，GC时从头到尾遍历数据，看到过期就淘汰，否则中断GC任务。由于bigcache按照FIFO先进先出存储，同时设置统一过期时间，因此ringbuffer的起始位置entry一定先过期。另外，ringbuffer环不会真正删除数据，只是移动head指针，未清理的字节数据会被新加入的数据覆盖。这里的GC其实就是初始化时，bigcache时起一个定时任务，定时清理过期的entry。

• 	if config.CleanWindow > 0 {
  		go func() {
  			ticker := time.NewTicker(config.CleanWindow)
  			defer ticker.Stop()
  			for {
  				select {
  				case <-ctx.Done():
  					fmt.Println("ctx done, shutting down bigcache cleanup routine")
  					return
  				case t := <-ticker.C:
                      // 定时清理
  					cache.cleanUp(uint64(t.Unix()))
  				case <-cache.close:
  					return
  				}
  			}
  		}()
  	}
  

• cleanUp会遍历每个分片，对每个shard执行cleanUp，利用相同的当前时间判断entry是否过期。

• func (c *BigCache) cleanUp(currentTimestamp uint64) {
  	for _, shard := range c.shards {
  		shard.cleanUp(currentTimestamp)
  	}
  }
  

• cacheShard.cleanUp会从head开始弹出所有过期的entry，调用removeOldestEntry清理过期数据。

• func (s *cacheShard) removeOldestEntry(reason RemoveReason) error {
      // 弹出entry,head指针右移
  	oldest, err := s.entries.Pop()
  	if err == nil {
          // 读取entry哈希值
  		hash := readHashFromEntry(oldest)
  		if hash == 0 {
  			// entry has been explicitly deleted with resetKeyFromEntry, ignore
  			return nil
  		}
          // 删除索引
  		delete(s.hashmap, hash)
          // 移除回调
  		s.onRemove(oldest, reason)
  		if s.statsEnabled {
  			delete(s.hashmapStats, hash)
  		}
  		return nil
  	}
  	return err
  }
  

• 

• 删除时机

  1.写操作时，ringbuffer已满又无法扩容时，删除旧数据(指针后移)，再写入新数据

  2.垃圾回收，从头开始就行过期判断，过期就清理