Geecache分布式缓存 | 第二天记录

主要内容：

单机并发缓存

1 Day2 单机并发缓存

1.1 互斥锁控制

当多个协程同时读写一个变量时，如果没有并发控制，很容易运行出与预期不一致的结果，甚至触发Panic。sync.Mutex是一个互斥锁，可以由不同的协程来加锁和解锁，从而保护并发下程序的安全性和可预期性。

go
var m sync.Mutex
var set = make(map[int]bool, 0)

func printOnce(num int) {
	m.Lock()
	if _, exist := set[num]; !exist {
		fmt.Println(num)
	}
	set[num] = true
	m.Unlock()
}

func main() {
	for i := 0; i < 10; i++ {
		go printOnce(100)
	}
	time.Sleep(time.Second)
}

另外，这个m.Unlock()还有另外一个地方可以去：

go
func printOnce(num int) {
	m.Lock()
	defer m.Unlock()
	if _, exist := set[num]; !exist {
		fmt.Println(num)
	}
	set[num] = true
}

defer关键字：defer和go一样，都是Go语言的那25个关键字之一。它的作用是释放资源，会在函数返回之前调用。如果有多个defer表达式，调用顺序类似于栈，越后面的defer表达式越先被调用。另外，defer也有个误区在：
go
func f() (result int) {
   defer func() {
       result++
   }()
   return 0
} // 实际返回1
它特别容易认为是返回0，类似的还有：
go
func f() (r int) {
     t := 5
     defer func() {
       t = t + 5
     }()
     return t
} // 实际返回5
go
func f() (r int) {
    defer func(r int) {
          r = r + 5
    }(r)
    return 1
} // 实际返回1
这个时候，认准这一条：返回的过程是先给返回值赋值，然后执行defer函数，最后回到调用该函数的地方。所以这三段分别可以被改写成这样：

1.2 支持并发读写

首先定义只读类ByteView，用于缓存值的抽象和封装，实现lru.go的Value接口：

go
// src/geecache/byteview.go
package geecache

// ByteView 只读数据结构 实现了Value接口 用于表示缓存的值 如果想要获取当前缓存的值 一律从GetByteCopy获取
type ByteView struct {
	cacheBytes []byte
}

// GetMemoryUsed 实现Value接口的方法 返回该缓存值的长度/占用内存多少
func (view ByteView) GetMemoryUsed() int {
	return len(view.cacheBytes)
}

// GetByteCopy 返回当前缓存值的一个拷贝 外部程序如果想要获取当前缓存的值 一律从该方法获取 用于防止缓存值被外部程序修改
func (view ByteView) GetByteCopy() []byte {
	return cloneBytes(view.cacheBytes)
}

// ToString 返回当前缓存的字符串表示
func (view ByteView) ToString() string {
	return string(view.cacheBytes)
}

func cloneBytes(b []byte) []byte {
	clone := make([]byte, len(b))
	copy(clone, b)
	return clone
}

之后再对昨天的LRU进行封装，并且追加并发保护：

go
// src/geecache/cache.go
package geecache

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

// cache 对LRU的一次封装 并且追加并发保护
type cache struct {
	mutex sync.Mutex // 互斥锁
	lru *lru.LRU // 封装的LRU
	cacheBytes int64
}

// add 对LRU.SetValue的封装
func (c *cache) add(key string, value ByteView) {
	c.mutex.Lock() // 互斥锁加锁
	defer c.mutex.Unlock()
	if c.lru == nil {
		c.lru = lru.NewLRU(c.cacheBytes, nil) // 懒加载
	}
	c.lru.SetValue(key, value)
}

// get 对LRU.GetValue的封装
func (c *cache) get(key string) (value ByteView, isOk bool) {
	c.mutex.Lock() // 互斥锁加锁
	defer c.mutex.Unlock()
	if c.lru == nil {
		return
	}

	if v, isOk := c.lru.GetValue(key); isOk {
		return v.(ByteView), isOk
	}
	return
}

1.3 主体结构Group

Group是该项目的核心数据结构，它负责和请求交互的部分。

在一个完整的后端服务中，缓存负责响应请求，如果未能命中请求的数据，则尝试从数据库中获取数据。该项目也是如此，暴露给请求的接口只有获取值的Get方法，由缓存自动根据被指定的方式添加新缓存，淘汰旧有缓存。

要想实现缓存自动根据被指定的方式获取新的缓存，首先先规定如何获取新的缓存。为此Group必须要有一个缓存未能命中时的回调函数：

go
// src/geecache/geecache.go
// Getter 接口 规定了一个Get方法 该方法用于规定缓存未命中时从哪里获得新的缓存
type Getter interface {
	Get(key string) ([]byte, error)
}

// GetterFunc 函数类型 专门用来实现Getter接口的函数类型
type GetterFunc func(key string) ([]byte, error)

// Get GetterFunc类下的 从Getter接口的Get函数实现而来的函数
func (function GetterFunc) Get(key string) ([]byte, error) {
	return function(key)
}

提示

这种，先有一个接口，再有一个函数类型，由这个函数类型实现这个接口的方法，叫做接口型函数。它的用处是，后续的函数如果将这个接口作为参数类型，则调用方可以传入实现这个接口的类，也可以直接传入方法。

之后是Group类的实现，注意Group类对外只有一个获取缓存值的方法。

go
// src/geecache/geecache.go
// Group 缓存对外交互的核心数据结构
type Group struct {
	name string // 该缓存的标识
	getter Getter // 缓存未能命中时的回调函数 类型是Getter接口
	mainCache cache // 缓存主体 是具有并发保护的LRU缓存
}

// 全局变量
var (
	mu sync.RWMutex // 互斥锁的高级版本 读写锁 在原有的锁功能上 增加读写特性 当读锁锁定时 只会阻止写 同理当写锁锁定时 会同时阻止读写
	groups = make(map[string]*Group) // 保存多个Group缓存
)

// NewGroup 构造函数
func NewGroup(name string, cacheBytes int64, getter Getter) *Group {
	if getter == nil {
		panic("nil getter")
	}
	mu.Lock()
	defer mu.Unlock()
	group := &Group{
		name: name,
		getter: getter,
		mainCache: cache{cacheBytes: cacheBytes},
	}
	groups[name] = group
	return group
}

// GetGroup 获取Group缓存
func GetGroup(name string) *Group {
	mu.RLock()
	g := groups[name]
	mu.RUnlock()
	return g
}

// GetFromCache 从缓存中获取值
func (g *Group) GetFromCache(key string) (ByteView, error) {
	if key == "" {
		return ByteView{}, fmt.Errorf("key is required")
	}
	if v, isOk := g.mainCache.get(key); isOk { // 缓存命中
		log.Println("[GeeCache] hit")
		return v, nil
	}
	// 缓存未命中 调用load函数
	return g.load(key)
}

// load 缓存未命中时 从别的地方加载缓存
func (g *Group) load(key string) (value ByteView, err error) {
	return g.getFromLocal(key)
}

// getFromLocal 从本地加载缓存 在这里调用Getter的Get函数 并且通过populateCache存入缓存
func (g *Group) getFromLocal(key string) (ByteView, error) {
	bytes, err := g.getter.Get(key)
	if err != nil {
		return ByteView{}, err
	}

	value := ByteView{cacheBytes: cloneBytes(bytes)}
	g.populateCache(key, value)
	return value, nil
}

// populateCache 新的缓存值 存入缓存
func (g *Group) populateCache(key string, value ByteView) {
	g.mainCache.add(key, value)
}

提示

读写互斥锁sync.RWMutex：互斥锁的高级版本。在原有的基础的加锁解锁功能外，增加了对读写的支持。如果读取锁是上锁状态，它不会阻止其他协程/线程的读取，但是会阻止写入行为，直到读取锁解锁；如果写入锁是上锁状态，它会阻止其他协程/线程的读取和写入行为。

2 今日内容结构图

3 测试文件

go
// src/geecache/geecache_test.go
package geecache

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

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", 2<<10, 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)
		}))

	for k, v := range db {
		if view, err := gee.GetFromCache(k); err != nil || view.ToString() != v {
			t.Fatal("failed to get value of Tom")
		} // load from callback function
		if _, err := gee.GetFromCache(k); err != nil || loadCounts[k] > 1 {
			t.Fatalf("cache %s miss", k)
		} // cache hit
	}

	if view, err := gee.GetFromCache("unknown"); err == nil {
		t.Fatalf("the value of unknow should be empty, but %s got", view)
	}
}