cache教程 2.单机并发缓存

0.对原教程的一些见解

个人认为原教程中两点知识的引入不够友好。

首先是只读数据结构 ByteView 的引入使用是有点迷茫的，可能不能很好理解为什么需要ByteView。

第二是主体结构 Group的引入也疑惑。其实要是熟悉groupcache，那对结构Group的使用是清晰明白的。而看该教程的人可能是没有了解过groupcache,直接就引入结构Group，可能不好理解。这一章节希望可以讲明白这两点。

1.统一的缓存的value对象

//该类型实现了NodeValue接口
type String string
 
func (d String) Len() int {
	return len(d)
}

在上节讲解中, 我们存入的每一个元素(键值对)都要计算大小。为了能计算大小，那存入缓存的 value 对象必须实现NodeValue接口的Len()方法。上一节的测试用例中存储的value对象是String（也即是string）。

那么问题来了, 我们存入的 value 可能是 string, int, 也可能自定义的结构体User等等。如果为每一种类型都实现一个 Len() 方法那确实是繁琐。因此，我们希望将存入的每个 value 都转化为统一的类型, 比如:字节数组 []byte。

我们可以抽象了一个只读数据结构 ByteView 用来表示缓存值。

ByteView 只有一个数据成员，b []byte，b 将会存储真实的缓存值。

b 是只读的，使用 ByteSlice() 方法返回一个拷贝，防止缓存值被外部程序修改。

//缓存值的抽象与封装
type ByteView struct {
	b []byte
}

func (v ByteView) Len() int {
	return len(v.b)
}

func (v ByteView) ByteSlice() []byte {
	return cloneByte(v.b)
}

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

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

2.实现缓存并发读写

上一节实现的LRU算法是不支持并发读写的。Go中map不是线程安全的。要实现并发读写map,需要加锁，可以使用sync.Mutex。

sync.Mutex 是一个互斥锁，可以由不同的协程加锁和解锁。

先回顾下上一节定义的缓存的整体数据结构

type Cache struct {
	maxBytes  int64      //允许的能使用的最大内存
	nbytes    int64      //已使用的内存
	ll        *list.List //双向链表
	cache     map[string]*list.Element
	OnEvicted func(key string, value NodeValue)
}

要是想的简单点，我们可以在该结构体Cache内部加上sync.Mutex并修改其方法的部分原有逻辑来实现并发读写。但这样就破坏了对扩展开放，对修改关闭的面向对象原则。这是不好的。

定义加锁的缓存对象

我们可以在Cache结构体基础上再封装一个可以支持并发读写的对象。

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

显然，该新对象中是需要有个互斥锁变量。而每个缓存对象都有能使用的最大内存量上限，使用cacheBytes 字段来存储这个值。

该cache对象也基于互斥锁和lru封装了 get 和 add 方法。

func (c *cache) add(key string, value ByteView) {
	c.mutex.Lock()
	defer c.mutex.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.mutex.Lock()
	defer c.mutex.Unlock()
	if c.lru == nil {
		return
	}

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

3.提升缓存并发读写能力

互斥锁引发的性能问题

引入锁之后，可能会引起性能问题，思考如下场景：

当有 A个线程访问库存的缓存数据时, 我们给 cache 对象加了锁，如果此时有 B个线程来访问商品缓存数据，这 A + B 个线程就需要共同竞争一把锁。

要是线程数量大的话，对性能是有影响的，那是因为所有的缓存都被一把锁把持住。那要是我们可以把缓存进行分组，这样首先就可以不用所有的线程都去抢一把锁了。

将缓存数据进行分组

为了提高缓存系统的并发读写的性能(降低锁的竞争程度)，我们想想是否可以再细分锁的范围，分段锁的设计。

可以理解成是先分段再锁，将原本的所有缓存分成了若干段，分别将这若干段放在了不同的组中，每个组有各自的锁，以此提高效率。

如此设计之后, 不同组的存缓数据就隔离了起来, 访问同一组数据的线程才会互相竞争。

这就引出了Group这个结构。

4.Group结构

定义一个分组结构，从上图也可知道，要去访问缓存，就需去找到该组，那如何辨别是这个组呢，这里就是通过组的名字去辨别的，每个组都有个名字。

// 紧接着我们定义一个 分组 类型
type Group struct {
    name      string // 分组名称
    mainCache cache  // 单个缓存对象
}

这时有多个组后，那如何通过组名字快速找到该组了？还是要用map。那肯定又涉及到多个线程并发读写 groups 。这里是找到对应组名字的组而加锁的。我们可以考虑用 读写锁 来解决这个问题。

这里使用读写锁应该比使用互斥锁可以提高并发度。

来看看创建组和通过名字获取组的函数

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

func NewGroup(name string, cacheBytes int64) *Group {
	rwMu.Lock()
	defer rwMu.Unlock()
	g := &Group{
		name:      name,
		mainCache: cache{cacheBytes: cacheBytes},
	}
	groups[name] = g
	return g
}

// 获取 Group 对象的方法
func GetGroup(name string) *Group {
	rwMu.RLock()
	defer rwMu.RUnlock()
	g := groups[name]
	return g
}

缓存查询回调方法

我们要考虑一种情况：如果缓存不存在，应从数据源（文件，数据库等）获取数据并添加到缓存中。

该Cache 是否应该支持多种数据源的配置呢？不应该，一是数据源的种类太多，没办法都实现；二是扩展性不好。如何从源头获取数据，应该是用户决定的事情，我们就把这件事交给用户好了。因此，我们设计了一个回调函数(callback)，在缓存不存在时，就可以调用该函数，得到源数据。

这个回调方法我们可以直接定义在上面的 Get 方法的入参中,也可以放在 Group 对象中，为了方便，我们放在Group内。

type Group struct {
    name      string // 组名
    mainCache cache  // 单个缓存对象
		// 新增回调函数
    getter    Getter

}

type Getter interface {
	Get(key string) ([]byte, error)
}

type GetterFunc func(key string) ([]byte, error)

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

函数类型实现某一个接口，称之为接口型函数，那么该函数也是接口。

其好处：当一个函数的参数类型是接口，那使用者在调用时既能够传入函数作为参数，也能够传入实现了该接口的结构体作为参数。

接口型函数不太理解的话，可以看Go接口型函数。

接口型函数在这章节的最后测试中也会进行讲解的，测试中有例子。

Group 的 Get 方法

首先从本地缓存中查找，若是有则直接返回该缓存数据即可。

若是缓存不存在(即是没击中)，则调用 load 方法，调用用户回调函数 g.getter.Get() 获取源数据，并且将源数据添加到缓存 mainCache 中。

func (g *Group) Get(key string) (ByteView, error) {
	if v, ok := g.mainCache.get(key); ok {
		return v, nil
	}
	return g.load(key)
}

func (g *Group) load(key string) (ByteView, error) {
	bytes, err := g.getter.Get(key)
	if err != nil {
		return ByteView{}, err
	}
	value := ByteView{b: cloneByte(bytes)}
	g.mainCache.add(key, value)    //将源数据添加到缓存mainCache
	return value, nil
}

至此，这一章节的单机并发缓存就已经完成了。

5.测试

// 缓存中没有的话，就从该db中查找
var db = map[string]string{
	"tom":  "100",
	"jack": "200",
	"sam":  "444",
}

// 统计某个键调用回调函数的次数
var loadCounts = make(map[string]int, len(db))

创建 group 实例，并测试 Get 方法。

主要测试了两种情况

1）在缓存为空的情况下，能够通过回调函数获取到源数据。
2）在缓存已经存在的情况下，是否直接从缓存中获取，为了实现这一点，使用 loadCounts 统计某个键调用回调函数的次数，如果次数大于1，则表示调用了多次回调函数，没有缓存。

func main() {
	//传函数入参    cache.GetterFunc(funcCbGet)是进行类型转换，不是执行函数
	cache := cache.NewGroup("scores", 2<<10, cache.GetterFunc(funcCbGet))
	//传结构体入参，也可以
	// cbGet := &search{}
	// cache := cache.NewGroup("scores", 2<<10, cbGet)

	for k, v := range db {
		if view, err := cache.Get(k); err != nil || view.String() != v {
			fmt.Println("failed to get value of ",k)
		}

		if _, err := cache.Get(k); err != nil || loadCounts[k] > 1 {
			fmt.Printf("cache %s miss", k)
		}
	}

	if view, err := cache.Get("unknown"); err == nil {
		fmt.Printf("the value of unknow should be empty, but %s got", view)
	}else {
		fmt.Println(err)
	}
}

// 函数的
func funcCbGet(key string) ([]byte, error) {
	fmt.Println("callback 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 exit", key)
}

// 结构体，实现了Getter接口的Get方法，
type search struct {
}

func (s *search) Get(key string) ([]byte, error) {
	fmt.Println("struct callback 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 exit", key)
}