使用Golang編寫Cache

code golang

學習如何使用golang從0編寫cache系統

前言

為什麼需要緩存?

  • 請求網頁時，網頁會將第一次請求的內容存入緩存中(通常為key-value)，以後再次訪問時，就可以直接從緩存中提取，從而降低載入時間

本地緩存 vs 分布式緩存

  • 本地緩存
    • 這邊的本地緩存通常指的為應用中的緩存組件，最大的優點就是應用和cache在同一個進程裡面，不會發生過多的網路開銷．缺點就是沒辦法共享緩存，個應用或集群都要維護自己的緩存，從而造成大量的開銷
  • 分布式緩存
    • 指的是應用和組件分離的緩存服務，優點就是緩存為一個獨立的應用，和本地分離，因此可以直接共享緩存

LRU緩存淘汰政策

  • 核心思想就是維護一個隊列，如果該條紀錄已經被訪問過了，就會到隊尾，那麼隊首變為最少訪問的紀錄，可以直接淘汰
  • 使用go語言中的container/list當作map中的值，該list為一個雙向隊列
  • entry是雙向隊列的數據，這麼做的目的是當要淘汰隊列的首節點時，使用key從字典中映射即可
type Cache struct {
	maxBytes  int64
	nbytes    int64
	ll        *list.List
	cache     map[string]*list.Element
	onEvicted func(key string, value Value)
}
 
type entry struct {
	key   string
	value Value
}
 
type Value interface {
	Len() int
}
 
func (c *Cache) Len() int {
	return c.ll.Len()
}
 
func New(maxBytes int64, onEvicted func(string, Value)) *Cache {
	return &Cache{
		maxBytes:  maxBytes,
		ll:        list.New(),
		cache:     make(map[string]*list.Element),
		onEvicted: onEvicted,
	}
}

查找功能

  • 如果鍵對應的值存在，將該節點移動到隊尾，並返回值
  • c.ll.MoveToFront(ele) 即為將值移動到隊尾(在這裡的front代表隊尾)
func (c *Cache) Get(key string) (value Value, ok bool) {
	if ele, ok := c.cache[key]; ok {
		c.ll.MoveToFront(ele)
		kv := ele.Value.(*entry)
		return kv.value, true
	}
	return
}

刪除功能

  • 實際上為淘汰最少訪問的首位節點，從linklist中刪除
  • 使用delete從字典中刪除該key
  • nbytes更新為當前正確的數值
  • 如果onEvicted不為nil的話，調用該函數
func (c *Cache) RemoveOldest() {
	ele := c.ll.Back()
	if ele != nil {
		c.ll.Remove(ele)
		kv := ele.Value.(*entry)
		delete(c.cache, kv.key)
		c.nbytes -= int64(len(kv.key)) + int64(kv.value.Len())
		if c.onEvicted != nil {
			c.onEvicted(kv.key, kv.value)
		}
	}
}

新增

  • 如果key存在的話，更新對應的值，並將該節點移動到隊尾
  • 不存在的話首先在隊尾新增該節點，並且新增和map的映射關係
  • 更新nbytes後如果超過最大值的話，調用Remove函數
func (c *Cache) Add(key string, value Value) {
	if ele, ok := c.cache[key]; ok {
		c.ll.MoveToFront(ele)
		kv := ele.Value.(*entry)
		c.nbytes -= int64(len(kv.key)) + int64(kv.value.Len())
		kv.value = value
	} else {
		ele := c.ll.PushFront(&entry{key, value})
		c.cache[key] = ele
		c.nbytes += int64(len(key)) + int64(value.Len())
	}
	for c.maxBytes != 0 && c.maxBytes < c.nbytes {
		c.RemoveOldest()
	}
}

分布式緩存算法

一至性哈希

  • 將key映射到2^32的空間中，將這個數字首尾相連，行程一個環
    • 計算節點/機器的哈希值，放置在環上
    • 計算key的哈希值，放置在環上，順時針尋找到的第一個節點，就是應選取的節點/機器

傾斜問題

  • 如果服務器節點過少，很容易都對應到同個節點，造成環存節點間負載不均衡
  • 使用虛擬節點的概念解決
    • 計算虛擬節點上的Hash值，放置在環上
    • 計算key的Hash值，在環上順時針尋找對應選取得虛擬節點例如peer2-1對應的就是真實節點peer2

實現一至性哈希

初始化

  • 定義函數類型的Hash，採取依賴注入的方式，允許用來替換成自定義的Hash函數，默認為ChecksumIEEE算法
  • Map是一至性哈希算法的主數據結構，包含
    1. Hash函數 hash
    2. 虛擬節點倍數 replicas
    3. 虛擬節點和真實節點的映射表 hashMap，key是虛擬節點的哈希值，值是真實節點的名稱
  • 構造函數New允許自定義虛擬節點倍數和Hash函數
type Hash func(data []byte) uint32
 
// Map constains all hashed keys
type Map struct {
	hash     Hash
	replicas int
	keys     []int // Sorted
	hashMap  map[int]string
}
 
// New creates a Map instance
func New(replicas int, fn Hash) *Map {
	m := &Map{
		replicas: replicas,
		hash:     fn,
		hashMap:  make(map[int]string),
	}
	if m.hash == nil {
		m.hash = crc32.ChecksumIEEE
	}
	return m
}

新增真實節點/機器

  • Add函數允許傳入0或多個真實節點的名稱
  • 對每一個真實節點key，對應創建m.replicas個虛擬節點
  • 使用m.hash()計算虛擬節點的哈希值，再使用append添加到環上
  • 在hashMap中增加虛擬節點和真實節點的映射關西
  • 依照環上哈希排序
func (m *Map) Add(keys ...string) {
	for _, key := range keys {
		for i := 0; i < m.replicas; i++ {
			hash := int(m.hash([]byte(strconv.Itoa(i) + key)))
			m.keys = append(m.keys, hash)
			m.hashMap[hash] = key
		}
	}
	sort.Ints(m.keys)
}

選擇節點

  • 計算key的哈希值
  • 順時針找到第一個匹配虛擬節點的下標idx，從m.keys中或去對應的哈希值，如果len相同，說明選擇第一個，因為keys是一個環的結構
  • 透過hashMap映射得到真實節點
func (m *Map) Get(key string) string {
	if len(m.keys) == 0 {
		return ""
	}
 
	hash := int(m.hash([]byte(key)))
	// Binary search for appropriate replica.
	idx := sort.Search(len(m.keys), func(i int) bool {
		return m.keys[i] >= hash
	})
 
	return m.hashMap[m.keys[idx%len(m.keys)]]
}

緩存問題和解決方案

  • 緩存雪崩
    • 緩存在同一時刻全部失效，遭成瞬間DB請求量大，壓力驟增引起雪崩．通常因為緩存服務器當機，緩存的key設置相同過期時間導致
  • 緩存擊穿
    • 一個存在的key在緩存過期的那一刻，同時有大量的請求，這些請求都會擊穿DB，造成順時DB請求量大，壓力驟增
  • 緩存穿透
    • 查詢一個不存在的數據，因為不存在則不會寫入緩存，所以美以都會請求DB，造成瞬間流量過大，穿透DB造成當機

實現解決方法

  • 創建call和group類型
  • call代表正在進行或已經結束的請求
  • Group管理不同key的請求
type call struct {
	wg  sync.WaitGroup
	val interface{}
	err error
}
 
type Group struct {
	mu sync.Mutex       // protects m
	m  map[string]*call
}
  • Do方法的作用為針對相同key，無論Do被調用多少次，函數fn都只會被調用一次，等待fn調用結束，return返回值或error
  • 使用延遲初始化g.m以提高內存使用效率
func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		g.mu.Unlock()
		c.wg.Wait()
		return c.val, c.err
	}
	c := new(call)
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()
 
	c.val, c.err = fn()
	c.wg.Done()
 
	g.mu.Lock()
	delete(g.m, key)
	g.mu.Unlock()
 
	return c.val, c.err
}

重點地方

  • g.mu是保護Group的成員變量m不被併發讀寫加上鎖
func (g *Group) Do(key string, fn func() (interface{}, error)) (interface{}, error) {
	if c, ok := g.m[key]; ok {
		c.wg.Wait()   //如果請求正在進行中則等待
		return c.val, c.err //請求結束返回結果 
	}
	c := new(call)
	c.wg.Add(1)       //發起請求前保護鎖
	g.m[key] = c      //添加g.m,表明key已經有對應的請求在處理
 
	c.val, c.err = fn() //調用fn發起請求
	c.wg.Done()         //請求結束
 
    delete(g.m, key)    //更新 g.m
    
	return c.val, c.err //返回结果
}

使用

  • 修改geecache中的Group，他加成員變量loader並更新構建函數NewGroup
  • 修改load函數，將原來的load邏輯使用g.loader.Do包裹，確保併發場景下針對同個key，load只會調用一次
type Group struct {
	name      string
	getter    Getter
	mainCache cache
	peers     PeerPicker
	// use singleflight.Group to make sure that
	// each key is only fetched once
	loader *singleflight.Group
}
 
func NewGroup(name string, cacheBytes int64, getter Getter) *Group {
    // ...
	g := &Group{
        // ...
		loader:    &singleflight.Group{},
	}
	return g
}
 
func (g *Group) load(key string) (value ByteView, err error) {
	// each key is only fetched once (either locally or remotely)
	// regardless of the number of concurrent callers.
	viewi, err := g.loader.Do(key, func() (interface{}, error) {
		if g.peers != nil {
			if peer, ok := g.peers.PickPeer(key); ok {
				if value, err = g.getFromPeer(peer, key); err == nil {
					return value, nil
				}
				log.Println("[GeeCache] Failed to get from peer", err)
			}
		}
 
		return g.getLocally(key)
	})
 
	if err == nil {
		return viewi.(ByteView), nil
	}
	return
}

使用Protocol通信

syntax = "proto3";
 
package geecachepb;
 
option go_package = "./";
 
message Request {
  string group = 1;
  string key = 2;
}
 
message Response {
  bytes value = 1;
}
 
service GroupCache {
  rpc Get(Request) returns (Response);
}
 
//protoc --go_out=. *.proto

使用

func (p *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    // ...
	// Write the value to the response body as a proto message.
	body, err := proto.Marshal(&pb.Response{Value: view.ByteSlice()})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/octet-stream")
	w.Write(body)
}
 
func (h *httpGetter) Get(in *pb.Request, out *pb.Response) error {
	u := fmt.Sprintf(
		"%v%v/%v",
		h.baseURL,
		url.QueryEscape(in.GetGroup()),
		url.QueryEscape(in.GetKey()),
	)
    res, err := http.Get(u)
	// ...
	if err = proto.Unmarshal(bytes, out); err != nil {
		return fmt.Errorf("decoding response body: %v", err)
	}
 
	return nil
}