0.序言

为什么需要 cache

cache 需要干什么

1.缓存淘汰策略

完整代码

version_1_LRU[geecache]
├── go.mod
└── lru
    ├── lru.go
    └── lru_test.go

lru/lru.go

package lru

import "container/list"

type Cache struct {
    maxBytes  int64                         // 最大的内存
    nBytes    int64                         // 已经被用作缓存的内存
    ll        *list.List                    // 双向链表
    cache     map[string]*list.Element      // 用 map 查找链表的节点
    OnEvicted func(key string, value Value) // 删除缓存节点的回调函数
}

// 链表节点的数据类型
type entry struct {
    key   string // 保存 key，删除链表缓存时，便于同步删除 map 中的 k/v 记录
    value Value  // 真正的缓存数据，允许 Value 是任何类型
}

type Value interface {
    Len() int // Len() 表示该数据的内存大小, 添加数据前,应该手动实现该数据类型的 Len 函数
}

func New(maxBytes int64, onEvicted func(key string, value Value)) *Cache {
    return &Cache{
        maxBytes:  maxBytes,
        ll:        list.New(),
        cache:     make(map[string]*list.Element),
        OnEvicted: onEvicted,
    }
}

// Add 增加缓存数据
func (c *Cache) Add(key string, value Value) {
    if _, ok := c.cache[key]; ok {
        c.Update(key, 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()
        }
    }
}

// RemoveOldest 删除链表尾部的缓存数据
// 用户并不需要调用，Add/Update时，可用空间不够，才会删除
func (c *Cache) RemoveOldest() {
    ele := c.ll.Back()
    if ele != nil {
        c.ll.Remove(ele)
        kv := ele.Value.(*entry) // 先获取链表节点的 Value 字段，类型是 any，并转化为 *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)
        }
    }
}

// Update 更新缓存数据
func (c *Cache) Update(key string, value Value) {
    if ele, ok := c.cache[key]; ok {
        c.ll.MoveToFront(ele)
        kv := ele.Value.(*entry)
        c.nBytes += int64(value.Len()) - int64(kv.value.Len())
        kv.value = value
    } else {
        c.Add(key, value)
    }
}

// Get 获取缓存数据
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
}

// Len 获取缓存数据的条数
func (c *Cache) Len() int {
    return c.ll.Len()
}

lru/lru_test.go

package lru

import (
    "fmt"
    "testing"
)

// 对于数据类型，需要进行封装，因为无法在原生类型 string 中添加 Len 方法
type String string

func (s String) Len() int {
    return len(s)
}

// TestFunction 测试增删改查基本功能
func TestFunction(t *testing.T) {
    lru := New(int64(10), nil)

    lru.Add("k1", String("1234"))
    fmt.Println(lru.Get("k1"))

    lru.RemoveOldest()
    fmt.Println(lru.Get("k1"))

    lru.Update("k2", String("123"))
    fmt.Println(lru.Get("k2"))
}

// TestAutoRemoveOldest 测试当使用内存超过 maxBytes，是否会触发 “无用” 节点的删除
func TestAutoRemoveOldest(t *testing.T) {
    testData := []struct {
        key   string
        value Value
    }{
        {"k1", String("1234567890")},
        {"k2", String("234567890")},
        {"k3", String("34567890")},
    }

    lru := New(int64(10), nil)

    for _, test := range testData {
        lru.Add(test.key, test.value)
    }

    for _, test := range testData {
        fmt.Println(lru.Get(test.key))
    }
}

// TestCallbackOnEvicted 测试删除缓存时回调函数是否能被调用
func TestCallbackOnEvicted(t *testing.T) {
    keys := make([]string, 0)
    lru := New(10, func(key string, value Value) {
        keys = append(keys, key)
    })
    lru.Add("k1", String("k1"))
    lru.Add("k2", String("k2"))
    lru.Add("k3", String("k3"))
    lru.Add("k4", String("k4"))

    fmt.Println(keys)
}

2.单机并发缓存

完整代码

version_2_singal_node[geecache]
├── geecache
│   ├── byteview.go  # 缓存值的抽象与封装
│   ├── cache.go  # 并发控制
│   ├── geecache.go  # 负责与用户交互，并拥有从外部数据源获取缓存并存储的功能
│   ├── geecache_test.go
│   └── lru  # 未更改，与上一节一致。缓存淘汰策略
│       ├── lru.go
│       └── lru_test.go
└── go.mod

byteview.go

package geecache

// 为什么要抽象 ByteView 数据类型来表示缓存值？
/* 1. 在 lru 中定义的 entry.value  是 Value 接口类型，所有传入的数据的类型均需要实现这个接口，也就是实现 Len 函数，比较麻烦。
      []byte 可以支持任何数据类型的存储，并为 []byte 抽象为 ByteView 类型，并实现 lru.Value 接口。

   2. 保证 ByteView 是只读类型
      如何保证？
      1. b 是小写，外部无法直接读取。
      2. 只能通过 ByteSlice 和 String 方法获取 b 的数据
        1.ByteSlice 返回 slice 时，会拷贝一个副本再返回
        2.String 将 b 的数据强制转换为 String。（外界也无法直接修改 b 的值）
*/

type ByteView struct {
    b []byte // slice
}

func (byteView ByteView) Len() int { // ByteView 实现 lru.Value 接口
    return len(byteView.b)
}

// 为什么要返回拷贝？
// 防止缓存值被外部修改，这里直接返回拷贝

func (byteView ByteView) ByteSlice() []byte {
    return cloneBytes(byteView.b) // 为什么不直接使用 make，要封装 cloneBytes 函数
}

// 为什么需要 cloneBytes ?
// 因为 []byte 是切片，传递时，不会深拷贝，传递的是视图，底层数据会被外界修改
func cloneBytes(b []byte) []byte {
    c := make([]byte, len(b))
    copy(c, b)
    return c
}

func (byteView ByteView) String() string {
    return string(byteView.b)
}

cache.go

package geecache

import (
    "sync"
    "geecache/geecache/lru"
)

// 为什么要封装 cache？
// 加锁，保证对 lru.Cache 的并发安全

type cache struct {
    mu         sync.Mutex
    lru        *lru.Cache
    cacheBytes int64
}

func (c *cache) add(key string, value ByteView) {
    c.mu.Lock()
    defer c.mu.Unlock()

    if c.lru == nil {
        c.lru = lru.New(c.cacheBytes, nil)
    }
    c.lru.Add(key, value)
}

func (c *cache) get(key string) (value ByteView, ok bool) {
    c.mu.Lock()
    defer c.mu.Unlock()

    if c.lru == nil {
        return
    }
    if v, ok := c.lru.Get(key); ok {
        return v.(ByteView), ok
    }

    return
}

geecache.go

package geecache

import (
    "fmt"
    "log"
    "sync"
)

// Getter 用于从各类的外部数据源获取数据
type Getter interface {
    Get(key string) ([]byte, error)
}

// GetterFunc 为了便于使用者传入匿名函数到 Getter 中
// 使用时，只需将匿名函数转化为 GetterFunc 类型，即可传入 Getter
type GetterFunc func(key string) ([]byte, error)

func (f GetterFunc) Get(key string) ([]byte, error) {
    return f(key)
}

// Group 负责与用户交互（获取缓存值），并拥有从外部数据源获取值并存储在缓存中的功能
type Group struct {
    name      string // 一个 Group 是一个命名空间，并有唯一的 name。比如可以创建多个 Group，储存不同类别的信息。
    getter    Getter // 当缓存未命中时，可以调用 Getter.Get 这个回调函数获取值，并储存在缓存中。
    mainCache cache  // 并发安全的缓存
}

var (
    groupMu sync.RWMutex
    groups  = make(map[string]*Group)
)

func NewGroup(name string, cacheBytes int64, getter Getter) *Group {
    if getter == nil {
        panic("nil Getter")
    }

    groupMu.Lock()
    defer groupMu.Unlock()

    g := &Group{
        name:      name,
        getter:    getter,
        mainCache: cache{cacheBytes: cacheBytes},
    }
    groups[name] = g
    return g
}

func GetGroup(name string) *Group {
    groupMu.RLock() // 为什么是读锁？
    g := groups[name]
    groupMu.RUnlock()
    return g
}

/*
                          是
接收 key --> 检查是否被缓存 -----> 返回缓存值 (1)
                |  否                         是
                |-----> 是否应当从远程节点获取 -----> 与远程节点交互 --> 返回缓存值 (2)
                            |  否
                            |-----> 调用`回调函数`，获取值并添加到缓存 --> 返回缓存值 (3)
Get 完成流程（1）
load + getLocally 完成流程(3)
*/

func (g *Group) Get(key string) (ByteView, error) {
    if key == "" {
        return ByteView{}, fmt.Errorf("key is required")
    }

    if v, ok := g.mainCache.get(key); ok {
        log.Println("[GeeCache] hit")
        return v, nil
    }

    // 缓存没有命中，就从其他数据源中获取
    return g.load(key)
}

// 当缓存不存在，Get -> load -> getLocally 获取数据
// 为什么不直接使用 getLocally，还要封装一个 load ？
// load 会先从 远程分布式节点获取，获取不到才会使用 getLocally。此处属于预留设计。
func (g *Group) load(key string) (ByteView, error) {
    return g.getLocally(key)
}

// getLocally 调用用户的回调函数 g.getter.Get，获取数据，并使用 populateCache 添加数据
func (g *Group) getLocally(key string) (ByteView, error) {
    bytes, err := g.getter.Get(key)
    if err != nil {
        return ByteView{}, err
    }
    value := ByteView{cloneBytes(bytes)}
    g.populateCache(key, value)
    return value, nil
}

// populateCache 将获取到的数据添加到缓存中
func (g *Group) populateCache(key string, value ByteView) {
    g.mainCache.add(key, value)
}

geecache_test.go

package geecache

import (
    "fmt"
    "log"
    "testing"
)

func TestGetter(t *testing.T) {
    var g Getter = GetterFunc(func(key string) ([]byte, error) {
        return []byte(key), nil
    })

    bytes, _ := g.Get("key")
    fmt.Printf("%c", bytes)
}

// 外部数据源
var db = map[string]string{
    "Tom":  "630",
    "Jack": "589",
    "Sam":  "567",
}

func TestGet(t *testing.T) {
    loadCounts := make(map[string]int, len(db)) // 记录加入缓存的次数

    gee := NewGroup("scores", 1024, GetterFunc(
        func(key string) ([]byte, error) { // 缓存未命中时，从其他数据源获取数据的函数
            log.Println("[SlowDB] search key", key)
            if v, ok := db[key]; ok {
                if _, ok := loadCounts[key]; !ok {
                    loadCounts[key] = 0
                }
                loadCounts[key] += 1
                return []byte(v), nil
            }
            return nil, fmt.Errorf("%s not exist", key)
        }))

    // 从缓存中查看各个 key，测试缓存从外部数据源获取数据的能力
    for k, v := range db {
        view, err := gee.Get(k)
        if err != nil {
            panic(err)
        }
        fmt.Println(k, v, view.String())
    }

    // 对于数据源中不存在的数据的处理
    view, err := gee.Get("unknown")
    if err != nil {
        fmt.Println(err)
    }
    fmt.Println(view.String())
}

3.HTTP 服务器

完整代码

version_3_http_server[geecache]
├── geecache
│   ├── byteview.go  # 未更改，与上一节一致。缓存值的抽象与封装
│   ├── cache.go  # 未更改，与上一节一致。并发控制
│   ├── geecache.go  # 未更改，与上一节一致。负责与用户交互，并拥有从外部数据源获取缓存并存储的功能
│   ├── geecache_test.go # 新增 http 测试
│   ├── http.go # 新增从 HTTP 服务器获取 cache
│   └── lru  # 未更改，与上一节一致。缓存淘汰策略
│       ├── lru.go
│       └── lru_test.go
└── go.mod

http.go

package geecache

import (
    "fmt"
    "log"
    "net/http"
    "strings"
)

const BASEPATH = "/_geecache/" // 请求路径应该是 "/<basepath>/<groupname>/<key>"

type HTTPPool struct {
    self     string
    basePath string
}

func NewHTTPPool(self string) *HTTPPool { // 为什么要设置这两个字段
    return &HTTPPool{
        self:     self,     // 本机的IP/端口
        basePath: BASEPATH, // 请求前缀，便于过滤请求
    }
}

func (h *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if !strings.HasPrefix(r.URL.Path, h.basePath) {
        panic("Error path: " + r.URL.Path)
    }

    h.Log("%s %s", r.Method, r.URL.Path)

    parts := strings.SplitN(r.URL.Path[len(BASEPATH):], "/", 2)
    if len(parts) != 2 {
        http.Error(w, "bad request", http.StatusBadRequest)
        return
    }

    groupName := parts[0]
    key := parts[1]
    group := GetGroup(groupName)
    if group == nil {
        http.Error(w, "no such group: "+groupName, http.StatusNotFound)
        return
    }
    view, err := group.Get(key)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "application/json")
    w.Write(view.ByteSlice())
}

func (h *HTTPPool) Log(format string, v ...interface{}) {
    log.Printf("[Server %s] %s", h.self, fmt.Sprintf(format, v...))
}

geecache_test.go

package geecache

import (
    "fmt"
    "log"
    "net/http"
    "testing"
)

// 外部数据源
var db = map[string]string{
    "Tom":  "630",
    "Jack": "589",
    "Sam":  "567",
}

func TestHTTP(t *testing.T) {
    NewGroup("scores", 1024, GetterFunc(
        func(key string) ([]byte, error) {
            log.Println("[SlowDB] search key", key)
            if v, ok := db[key]; ok {
                return []byte(v), nil
            }
            return nil, fmt.Errorf("%s not exist", key)
        }))

    addr := "localhost:9999"
    peers := NewHTTPPool(addr)
    log.Println("cache is running at", addr)
    log.Fatal(http.ListenAndServe(addr, peers))
}

/*
curl http://localhost:9999/_geecache/scores/Tom
curl http://localhost:9999/_geecache/scores/xxx

log:
2023/10/16 17:40:24 [Server localhost:9999] GET /_geecache/scores/Tom
2023/10/16 17:40:24 [SlowDB] search key Tom
2023/10/16 17:40:51 [Server localhost:9999] GET /_geecache/scores/567
2023/10/16 17:40:51 [SlowDB] search key 567
*/

4.一致性哈希

完整代码

version_4_consistent_hashing[geecache]
├── geecache
│   ├── byteview.go  # 未更改，与上一节一致。缓存值的抽象与封装
│   ├── cache.go  # 未更改，与上一节一致。并发控制
│   ├── consistenthash
│   │   └── consistent_hash.go # 新增一致性哈希
│   ├── geecache.go  # 未更改，与上一节一致。负责与用户交互，并拥有从外部数据源获取缓存并存储的功能
│   ├── geecache_test.go
│   ├── http.go # 未更改，与上一节一致。HTTP 服务
│   └── lru
│       ├── lru.go
│       └── lru_test.go
└── go.mod

consistenthash/consistent_hash.go

package consistenthash

import (
    "hash/crc32"
    "sort"
    "strconv"
)

type Hash func(data []byte) uint32 // hash 函数类型

type Map struct { // 一致性哈希算法的主要数据结构
    hash     Hash           // 设置自定义一种哈希算法函数
    replicas int            // 虚拟节点的倍数
    keys     []int          // 哈希环
    hashMap  map[int]string // 虚拟节点和真实节点的映射表，key 是虚拟节点的哈希值，value 是真实节点的名称
}

func New(replicas int, fn Hash) *Map {
    m := &Map{
        hash:     fn,
        replicas: replicas,
        hashMap:  make(map[int]string),
    }
    if m.hash == nil { // 默认使用 crc32.ChecksumIEEE 哈希算法
        m.hash = crc32.ChecksumIEEE
    }
    return m
}

// Add 添加真实节点
func (m *Map) Add(keys ...string) {
    for _, key := range keys { // 对于每个真实节点，添加 m.replicas 个虚拟节点。
        for i := 0; i < m.replicas; i++ {
            hash := int(m.hash([]byte(strconv.Itoa(i) + key))) // 计算虚拟节点的 hash 值
            m.keys = append(m.keys, hash)                      // 虚拟节点加入环
            m.hashMap[hash] = key                              // 将虚拟节点和真实节点加入映射表
        }
    }
    sort.Ints(m.keys) // 对虚拟节点排序
}

// Get 选择节点
func (m *Map) Get(key string) string {
    if len(m.keys) == 0 {
        return ""
    }

    hash := int(m.hash([]byte(key)))

    // 二分查找虚拟节点
    index := sort.Search(len(m.keys), func(i int) bool { // 这个函数的意义
        return m.keys[i] >= hash
    })

    return m.hashMap[m.keys[index%len(m.keys)]]
}