主要内容：

1. 一致性哈希原理
2. 一致性哈希实现



### 1 一致性哈希原理

对于分布式缓存来说，自己接收到请求后，如果在本地查找缓存找不到，下一步肯定是试图请求远程节点来获得缓存，那么假设现在整个分布式缓存算上本节点一共有10个节点，那么本节点在试图请求远程节点时应该请求谁？随机访问肯定不行，有可能节点1访问了节点2，然后发现节点2也没有缓存，再访问节点3，发现节点3也没有缓存……，可见随机访问的效率一定很低。

自定义哈希呢？比如说根据请求的key来计算哈希然后对10求模，求得的数字作为访问节点的依据。这样确实也可以，但是它仍然有一个问题：缓存雪崩。如果这10个节点中有一个节点失效，或者有一个新的节点加入分布式缓存，那么自定义哈希求模就是对9或者11求模了，绝大部分的缓存都会失效，自然会引发缓存雪崩。

一致性哈希就是解决普通哈希的引发缓存雪崩的隐患替代方案。它的大致原理是将请求的key值映射到一个环上，与其同时环上存储着分布式缓存系统的10个节点。确定了key的映射值之后就在环上往后找，找到的第一个节点就是要远程请求的节点。如果这10个节点中有节点失效了，那么请求也会顺延到下一个节点上，而不会引发缓存雪崩。

![110f0321e2537b85bd021da7a1bf59e.jpg](https://www.misaka19327.cc/static/img/28a354ec6f720b494f430348b06fc917.110f0321e2537b85bd021da7a1bf59e.jpg)

但是一致性哈希也有自己的问题：数据倾斜。还是那个环，还是那10个节点，但是如果这10个节点都只在环的半侧，另外半侧没有节点，这就会造成第一个节点极其繁忙，其余的节点浪费资源。缓解数据倾斜的方法是虚拟节点：为一个节点映射多个虚拟节点，这样减少了节点之间的距离的同时，资源消耗也少：只需要维护节点到虚拟节点的映射关系即可。

![4da1ab8531109accda801992ac83996.jpg](https://www.misaka19327.cc/static/img/995c87adedee2ed99d9a03cb6c2aed35.4da1ab8531109accda801992ac83996.jpg)

### 2 Day4 一致性哈希实现

先明确在Go里实现一致性哈希要维护什么：

1. 哈希函数，用来映射值到环上的
2. 一个数组，用来表示环
3. 一个字典，用来存储虚节点到节点的映射
4. 一个倍数，用来表示一个节点有几个虚节点

一致性哈希能做什么：

1. 添加新的节点
2. 根据传入的key来返回对应的真实节点

``` go
// src/geecache/consistenthash/consistenthash.go
package consistenthash

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

// HashFunc 哈希函数类型
type HashFunc func(data []byte) uint32

// Map 一致性哈希的主要数据结构 这里的一致性哈希维护的节点仅为节点名
type Map struct {
	hash HashFunc
	replicasNumber int // 真实节点和虚拟节点的映射倍数
	keys []int // 一致性哈希的环的抽象版
	hashmap map[int]string // 虚拟节点到真实节点的映射
}

// NewMap 一致性哈希的构造函数 默认情况下选择CRC32校验和作为哈希值
func NewMap(hashFunc HashFunc, replicasNumber int) (result *Map) {
	result = &Map{
		hash: hashFunc,
		replicasNumber: replicasNumber,
		hashmap: make(map[int]string),
	}
	if hashFunc == nil {
		result.hash = crc32.ChecksumIEEE
	}
	return
}

// AddRealNode 为一致性哈希添加真实节点（可以一次添加多个真实节点）
func (m *Map) AddRealNode(keys ...string) {
	for _, key := range keys {
		for i := 0; i < m.replicasNumber; i++ {
			hash := int(m.hash([]byte(strconv.Itoa(i) + key))) // 这个strconv.Itoa等效FormatInt 从整型转字符串
			m.keys = append(m.keys, hash)
			m.hashmap[hash] = key // 添加虚拟节点到真实节点的映射
		}
	}
	sort.Ints(m.keys) // 排序
}

// GetRealNodeByKey 根据key来获得真实节点
func (m *Map) GetRealNodeByKey(key string) (result string) {
	if len(m.keys) == 0 { // key检查是否有效
		return ""
	}
	
	keyHash := int(m.hash([]byte(key)))
	index := sort.Search(len(m.keys), func(i int) bool { // 二分查找 如果找不到符合条件的就返回len(m.keys)
		return m.keys[i] >= keyHash
	})
	
	return m.hashmap[m.keys[index % len(m.keys)]] // 去映射里查找真实节点
}
```

注意代码实现的时候并没有真的实现那个环，而是用一个数组代替。这个数组存储着所有真实节点和其虚拟节点的哈希值，并且调用AddRealNode后都会对该数组进行排序。当传入key后，以二分查找的方式寻找比传入的key的哈希更大的哈希，最后根据找到的下标去keys里找到哈希，再根据哈希去hashmap里找到真实节点。

> sort.Search函数，它接收两个参数。第一个是要查找的范围，第二个是一个判断函数，Search函数会根据该函数决定是否继续查找。如果从始至终没有找到一个位置使得判断函数为真，则返回查找范围作为索引。

### 3 测试文件

``` go
package main

import (
	"GeeCache/src/geecache/consistenthash"
	"strconv"
	"testing"
)

func TestHash(t *testing.T) {
	test_map := consistenthash.NewMap(func(data []byte) uint32 {
		i, _ := strconv.Atoi(string(data))
		return uint32(i)
	}, 3)
	
	test_map.AddRealNode("2", "4", "6")
	
	test_case := map[string]string {
		"2": "2",
		"11": "2",
		"23": "4",
		"27": "2",
	}
	
	for k,v := range test_case {
		if r := test_map.GetRealNodeByKey(k); r != v {
			t.Errorf("Asking for %s, should have yielded %s, actually is %s", k, v, r)
		}
	}
	
	test_map.AddRealNode("8")
	test_case["27"] = "8"
	for k,v := range test_case {
		if r := test_map.GetRealNodeByKey(k); r != v {
			t.Errorf("Asking for %s, should have yielded %s, actually is %s", k, v, r)
		}
	}
}
```

GeeCache分布式缓存 | 第四天记录

主要内容：

1. HTTP服务端实现



### 1 HTTP服务端示例

Go语言可以很轻松地实现一个HTTP服务端，示例如下，下例创建了一个在19326端口接收任何HTTP请求的HTTP服务端。

``` go
package main

import (
	"log"
	"net/http"
)

type server int // server类 所有要传到http包里作为Handler的类必须实现ServeHTTP方法

func main() {
	var s server
	http.ListenAndServe("127.0.0.1:19326", &s)
}

// ServeHTTP 自己实现的ServeHTTP方法 在这个类中记录了HTTP请求的路径 并且响应Hello World
func (h *server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	log.Println(r.URL.Path)
	_, err := w.Write([]byte("Hello World!"))
	if err != nil {
		return 
	}
}
```

效果如下：

![image.png](https://www.misaka19327.cc/static/img/4bd0b3dcbf2eb57967b5924c10029c05.image.png)

![image.png](https://www.misaka19327.cc/static/img/61db91d8760845376650bb4dfc84ef33.image.png)

### 2 Day3 HTTP服务端

得益于已经对缓存进行了有效的并发保护，在HTTP服务端这里直接调用函数即可。

``` go
// src/geecache/http.go
package geecache

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

const defaultBasePath = "/_geecache/" // 默认资源地址

// HTTPPool 一个缓存对应一个HTTP池 记录自身的地址和URL
type HTTPPool struct {
	selfAddr string // 记录缓存自身的地址 包括端口
	basePath string // 记录URL
}

// NewHTTPPool HTTPPool的构造方法
func NewHTTPPool(selfAddr string) (result *HTTPPool) {
	result = &HTTPPool{
		selfAddr: selfAddr,
		basePath: defaultBasePath,
	}
	return
}

// Log 记录信息 参数v可传多个值 这些值会按format来进行格式化 再进入log
func (pool *HTTPPool) Log(format string, v ...interface{}) {
	log.Printf("[Server %s] %s", pool.selfAddr, fmt.Sprintf(format, v...))
}

// ServeHTTP HTTP服务端的Handler方法 对有效的缓存请求进行响应
func (pool *HTTPPool) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if !strings.HasPrefix(r.URL.Path, pool.basePath) { // 检查请求是否有效
		panic("HTTPPool is serving unexpected path: " + r.URL.Path)
	}
	pool.Log("%s %s", r.Method, r.URL.Path) // log记录该次请求的信息
	
	parts := strings.SplitN(r.URL.Path[len(pool.basePath):], "/", 2) // 切片 按'/'切两段
	if len(parts) != 2 { // 检查请求是否有效
		http.Error(w, "bad request", http.StatusBadRequest) // 400
		return
	}
	
	groupName := parts[0]
	key := parts[1]
	
	group := GetGroup(groupName)
	if group == nil { // 请求的缓存不存在
		http.Error(w, "no such group:" + groupName, http.StatusNotFound) // 404
		return
	}
	view, err := group.GetFromCache(key)
	if err != nil { // 缓存请求失败
		http.Error(w, err.Error(), http.StatusInternalServerError) // 500
		return
	}
	
	w.Header().Set("Content-Type", "application/octet-stream")
	_, err = w.Write(view.GetByteCopy())
	if err != nil { // 写入响应失败
		http.Error(w, "write into responseWriter error: " + err.Error(), http.StatusInternalServerError) // 500
		return
	}
}
```

### 3 今日内容结构图

![3b8b6fc18b75ba1bd79bc5909cce66b.jpg](https://www.misaka19327.cc/static/img/49163fd044aef33c63ea1741193f8406.3b8b6fc18b75ba1bd79bc5909cce66b.jpg)

### 4 测试文件

``` go
package main

import (
	"GeeCache/src/geecache"
	"fmt"
	"log"
	"net/http"
)

var db1 = map[string]string{
	"Tom":  "630",
	"Jack": "589",
	"Sam":  "567",
}

func main() {
	geecache.NewGroup("scores", 2<<10, geecache.GetterFunc(
		func(key string) ([]byte, error) {
			log.Println("[SlowDB] search key", key)
			if v, ok := db1[key]; ok {
				return []byte(v), nil
			}
			return nil, fmt.Errorf("%s not exist", key)
		}))
	
	addr := "localhost:9999"
	peers := geecache.NewHTTPPool(addr)
	log.Println("geecache is running at", addr)
	log.Fatal(http.ListenAndServe(addr, peers))
}
```

测试效果：

![image.png](https://www.misaka19327.cc/static/img/8797cb6beee3129f9d6d2d29871961ce.image.png)

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

主要内容：

1. 单机并发缓存



### 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函数，最后回到调用该函数的地方。所以这三段分别可以被改写成这样：
> 
> ![image.png](https://www.misaka19327.cc/static/img/9ece2bb4500371164f6e7b6639905e84.image.png)
> 
> ![image.png](https://www.misaka19327.cc/static/img/4ea84e805512605418030d1909373d5d.image.png)
>
> ![image.png](https://www.misaka19327.cc/static/img/7802933970d75cf82b19ad32b3d62cec.image.png)

##### 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)
}
```

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

之后是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)
}
```

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

### 2 今日内容结构图

![41fa869e008ea7d8f5163917a4be585.jpg](https://www.misaka19327.cc/static/img/630c3931dec88feb8c636445ca400b09.41fa869e008ea7d8f5163917a4be585.jpg)

### 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)
	}
}
```

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

主要内容：

1. 序言部分
2. LRU缓存策略实现



### 1 序言部分记录

缓存对于后端来说是必要的，毕竟数据库操作很耗时，不能每个请求都去读写数据库，会严重拖慢服务的速度。缓存可以将近期读取过的数据存入缓存，每次请求数据时，如果命中缓存即可直接返回缓存值。

一个分布式缓存要考虑的问题：

1. 缓存淘汰策略：在缓存已满时，如何淘汰旧有缓存才能保证其余缓存在接下来的请求中命中率最高？
2. 并发控制：多个请求同时到达缓存，如何设计并发控制才能保证线程安全的同时速度最快？
3. 多节点通信同步：如果单机缓存已经到达性能极限，如何扩展缓存系统？如果是水平扩展，节点之间如何通信？如果是垂直扩展，扩展策略应该如何设计？
4. 读写特性：缓存是否为只读？缓存是否设计淘汰时间？缓存类型是什么？

:::tip{title="提示"}
一提到缓存，三个和缓存相关的问题是必须要提的，即缓存雪崩、缓存击穿和缓存穿透。

- **缓存雪崩**：缓存内的大量数据的过期淘汰时间相近，使得这些数据被淘汰后，缓存未能命中大量的请求（或者缓存自身由于某种原因崩溃了），这些请求直接到了数据库里，直接导致数据库崩溃。

- **缓存击穿**：假设缓存中有一个值被极其频繁地访问（该值一般被称为**热点数据**），当这个值过期后，大量的访问请求到达数据库，导致数据库崩溃。

- **缓存穿透**：如果用户访问的数据既不在缓存里，也不在数据库里，后台就无法构建缓存来响应请求。如果大量的这种目标数据既不在缓存里也不在数据库里的请求到来，这些请求一样会到达数据库，导致数据库崩溃。

这三种缓存异常，都是因为各种原因导致缓存命中率不高，使得大量的请求压垮数据库。这些缓存异常的常见解决方法如下：

- **缓存雪崩**：
    1. 均匀设置过期时间：在添加缓存时，在过期时间处添加一个随机数，使得多个缓存不要在同一时间内淘汰
    2. 互斥锁：在构建缓存时上锁。如果有请求到来，并且缓存未能命中，就先访问锁，如果是未上锁的状态，就正常构建缓存；如果是上锁状态，就等锁结束后再次访问缓存，确认是否命中。（另：实现互斥锁的时候，最好设置超时时间，不然第一个请求拿到了锁，然后这个请求发生了某种意外而一直阻塞，一直不释放锁，这时其他请求也一直拿不到锁，整个系统就会出现无响应的现象。）
    3. 后台多一个线程负责更新缓存，请求只到达缓存，不到数据库。如果缓存未命中，就通过消息队列到后台线程去更新缓存。
    4. 如果缓存自己崩了，可以启动服务熔断/请求限流或者构建缓存集群来解决。
 
- **缓存击穿**：同缓存雪崩2，3
- **缓存穿透**：
    1. 限制非法请求
    2. 缓存空值：如果发现数据库没有数据，可以在缓存中对这次请求加一个空值缓存，避免每次都要请求数据库。
    3. 布隆过滤器
:::

Geecache项目实现日程安排如下：

1. LRU缓存淘汰策略
2. 单机并发缓存
3. HTTP客户端
4. 一致性哈希
5. 分布式节点
6. 防止缓存击穿
7. 使用Protubuf通信

另外，该项目可以考虑的扩展方向：

- 多个淘汰策略，比如LFU、ARC
- HTTP通信改为RPC通信
- 细化锁的粒度来提高并发性能
- 实现热点互备来避免热点数据频繁请求影响性能
- 加入etcd
- 加入缓存过期机制

### 1 Day1 LRU缓存策略实现

常见的缓存策略有FIFO、LRU、LFU、ARC等，该项目使用LRU作为缓存策略。

> FIFO算法，顾名思义，先进先出，当缓存满时会自动删除最先进入缓存的数据。但是每次被删除的数据也有可能是最常访问的数据，这一模式并不能很好地匹配缓存的需求。

> LFU算法（Least Frequently Used）：最少使用算法，每次缓存满时会删除最久未使用过的缓存。这种情况能够很好地满足缓存的需求，但是它所占用的开销算是很大的：不仅要存键值对缓存，还要存这个缓存对应的访问次数，而且每次删除缓存时都需要对访问次数排序。不仅如此，如果对缓存的访问形式突然发生改变，以前经常访问的缓存以后都不再访问，那么要淘汰以前访问次数多的缓存也需要很长的时间。

> LRU算法（Least Recently Used）最近最少使用，相对于仅考虑时间的FIFO和仅考虑频率的LFU，该算法认为，如果一个数据被访问了，那么它在未来的一段时间访问的概率就会很高。

LRU算法在Go中的实现，需要一个字典和一个队列（双向链表Cosplay的）组成。队列存储缓存被访问的次序，字典存储具体的数据。每次删除数据时，就将队首的元素出队；每次有缓存被访问时，就将其重新入队，回到队尾。

> GoLand的tab默认是4空格，所以看着可能有点怪。

``` go
// src/geecache/lru/lru.go
// LRU 采用了LRU策略的缓存类
type LRU struct {
	maxMemoryBytes  int64                         // 允许使用的最大内存
	memoryUsedBytes int64                         // 当前已经使用的内存
	queue           *list.List                    // 队列 存储访问频率
	cacheMap        map[string]*list.Element      // 字典 存储具体的缓存的键值对 这里的值存的是双向链表的元素指针类型 具体的值在链表里
	OnEntryDeleted  func(key string, value Value) // 当缓存被删除时的回调函数 这种函数类型的默认值就是nil
}

// NewLRU LRU的构造函数
func NewLRU(maxMemoryBytes int64, onEntryDeleted func(string, Value)) (cache *LRU) {
	cache = &LRU{
		maxMemoryBytes: maxMemoryBytes,
		queue:          list.New(),
		cacheMap:       make(map[string]*list.Element),
		OnEntryDeleted: onEntryDeleted,
	}
	return
}

// Value 一条键值对缓存具体要存储的值的接口 该接口的目的是为了确保值对所有类型的通用性
type Value interface {
	GetMemoryUsed() int // 返回该缓存的值所占用的内存大小
}

// entry Cache中双向链表所存储的类型
type entry struct {
	key string
	// 之所以在双向链表中存储键 是为了删除更加方便 能够直接根据键去字典中删除元素
	value Value // 值必须是实现了Value接口的类型
}

// 开始实现功能

// GetValue 在缓存里查找值
func (cache *LRU) GetValue(key string) (value Value, isOk bool) {
	if element, isOk := cache.cacheMap[key]; isOk {
		cache.queue.MoveToFront(element) // 如果元素存在 则移动到队尾 注意双向链表的队尾是相对的 规定Front是队尾即可
		entry := element.Value.(*entry)  // 这个节点的值取出来 转换为entry类型
		return entry.value, true
	}
	value = nil
	isOk = false
	return
}

// RemoveOldestCache 淘汰一次缓存
func (cache *LRU) RemoveOldestCache() {
	element := cache.queue.Back() // 取出队首元素
	if element != nil {
		cache.queue.Remove(element) // 从双向链表中移除该元素
		cacheEntry := element.Value.(*entry)
		delete(cache.cacheMap, cacheEntry.key) // 从字典中移除该键值对
		cache.memoryUsedBytes -= int64(len(cacheEntry.key)) + int64(cacheEntry.value.GetMemoryUsed()) // 修改已使用的内存
		if cache.OnEntryDeleted != nil { // 如果回调函数不为空 则调用回调函数
			cache.OnEntryDeleted(cacheEntry.key, cacheEntry.value)
		}
	}
}

// SetValue 添加/修改缓存
func (cache *LRU) SetValue(key string, value Value) {
	if element, isOk := cache.cacheMap[key]; isOk { // 键存在
		cache.queue.MoveToFront(element)
		cacheEntry := element.Value.(*entry)
		cache.memoryUsedBytes += int64(value.GetMemoryUsed()) - int64(cacheEntry.value.GetMemoryUsed())
		cacheEntry.value = value
	} else { // 键不存在
		element = cache.queue.PushFront(&entry{key: key, value: value})
		cache.cacheMap[key] = element
		cache.memoryUsedBytes += int64(len(key)) + int64(value.GetMemoryUsed())
	}
	for cache.memoryUsedBytes > cache.maxMemoryBytes && cache.memoryUsedBytes != 0 { // 看已经使用的缓存内存有多大来淘汰旧缓存
		cache.RemoveOldestCache()
	}
}

// GetLRUEntryNumber 返回当前已缓存的键值对数量
func (cache *LRU) GetLRUEntryNumber()(len int) {
	len = cache.queue.Len()
	return
}
```

### 3 今日内容结构图

![7a8850f09e60bcdd8e85183db6b6124.jpg](https://www.misaka19327.cc/static/img/0614c00e47e6238a1bc3fe9f3437f87e.7a8850f09e60bcdd8e85183db6b6124.jpg)

### 4 测试文件

``` go
// src/geecache/lru/lru_test.go
package lru

import (
	"reflect"
	"testing"
)

type String string

func (string String) GetMemoryUsed() int { // 要实现Value接口的那个方法 下面才能直接添加什么的
	return len(string)
}

func TestCache_GetValue(t *testing.T) {
	lru := NewLRU(int64(100), nil)
	lru.SetValue("key1", String("1234"))
	if value, isOK := lru.GetValue("key1"); !isOK || value != String("1234") {
		t.Fatalf("cache hit key1=1234 failed, value actually is %s", value)
	}
	if _, isOK := lru.GetValue("key2"); isOK {
		t.Fatalf("cache miss key2 failed")
	}
}

func TestCache_RemoveOldestCache(t *testing.T) {
	key1, key2, key3 := "key1", "key2", "key3"
	value1, value2, value3 := "value1", "value2", "value3"

	cap := len(key1 + key2 + value1 + value2)
	lru := NewLRU(int64(cap), nil)
	lru.SetValue(key1, String(value1))
	lru.SetValue(key2, String(value2))
	lru.SetValue(key3, String(value3))

	if _, isOk := lru.GetValue(key1); isOk || lru.GetLRUEntryNumber() != 2 {
		t.Fatalf("Removeoldest key1 failed")
	}
}

func TestCache_OnEntryDeleted (t *testing.T) {
	keys := make([]string, 0)
	callback := func(key string, value Value) {
		keys = append(keys, key)
	}
	lru := NewLRU(int64(20), callback)
	lru.SetValue("key1", String("value1"))
	lru.SetValue("key2", String("value2"))
	lru.SetValue("key3", String("value3"))
	lru.SetValue("key4", String("value3"))
	expected := []string {"key1", "key2"}

	if !reflect.DeepEqual(expected, keys) {
		t.Fatalf("Call OnEvicted failed, expect keys equals to %s, now keys are %s", expected, keys)
	}
}
```

Geecache分布式缓存 | 序言&第一天记录

今日任务：

1. 01背包问题
2. 01背包问题+滚动dp数组
3. LeetCode416 分割等和子集



资料来源：

1. [代码随想录 | 01背包问题](https://programmercarl.com/%E8%83%8C%E5%8C%85%E7%90%86%E8%AE%BA%E5%9F%BA%E7%A1%8001%E8%83%8C%E5%8C%85-1.html)
2. [代码随想录 | 01背包问题+滚动dp数组](https://programmercarl.com/%E8%83%8C%E5%8C%85%E7%90%86%E8%AE%BA%E5%9F%BA%E7%A1%8001%E8%83%8C%E5%8C%85-2.html)
3. [代码随想录 | LeetCode416 分割等和子集](https://programmercarl.com/0416.%E5%88%86%E5%89%B2%E7%AD%89%E5%92%8C%E5%AD%90%E9%9B%86.html)

### 1 01背包问题

背包问题，指的是一类可以被抽象为以下问题的问题类型：

> 假设当前有一个容量为j的背包，面前有n个体积、价值各不相同的物品，要如何挑选物品使得一次能带走的所有物品的价值最大？

背包问题可以被细分为多个小问题：01背包（体积和价值相同的物品只有一个）、多重背包（体积和价值相同的物品有多个，且不同的物品数量不同）、完全背包（体积和价值相同的物品可以无限取）和分组背包（物品按组打包，每组物品选一个）等等。

其中，01背包是最基础的背包问题，它可以被阐述为：

> 假设当前有一个容量为j的背包，面前有n种体积、价值各不相同的物品，**且每种物品只有1个**，要如何挑选物品使得一次能带走的所有物品的价值最大？

首先先说暴力解法。对，01背包是有暴力解法的（不如说整个背包问题都可以暴力解），它的暴力解法就是回溯。既然每个物品只有取和不取两种选择，那么直接对每种物品的取和不取两种情况向下搜索，然后再回溯即可。只是这种解法在一般情况下都会超时，因为它的时间复杂度是$$ 2 ^ n$$。

再说动规解法，它的核心递推思路就是：对每个物品，对取这个物品和不取这个物品的两种情况的背包总价值取最大值，而这两种情况可以由**判断这个物品前背包的最大价值**和**背包容量减去当前物品重量所对应的背包容量下的背包最大价值加上这个物品的价值**得到。假设现在有一个容量为4的背包，有三个物品，第一个物品重量1价值15、第二个物品重量3价值20、第三个物品重量4价值30，那么它的dp数组应该长这样：

![21b4b58e1ce6f28000fb9936e52e7f5.jpg](https://www.misaka19327.cc/static/img/9f1a70f38b7b1287cf811e938ebafa87.21b4b58e1ce6f28000fb9936e52e7f5.jpg)

红字即为那一种情况的递推过程。

对着这个图，还是动态规划五部曲：

1. 确定dp数组和下标含义：对于这个dp数组，在它下标为i，j的位置的含义是，背包容量为j时，任取物品1到i所能携带的最大价值
2. 递推公式：没有取到物品i时，背包容量为j的最大价值和没有取到物品i时，背包容量为j减去物品i的重量的最大价值，再加上物品i的价值的最大值（具体看红色箭头）
3. dp数组初始化：只初始化背包容量为0时，对于所有物品的可以携带的最大价值和只取物品1时，所有背包容量情况下的最大价值（即为图中的第一排和第一列），其他位置无需初始化（反正都是要在之后的递推中被覆盖的）
4. 确定遍历顺序：一定是从左向右、从上到下遍历（因为递推出一种情况所需要的值，**一定是从它的上一排的左侧和正上方取值**，所以遍历时一定保证上一排的左侧和正上方一定有值），至于是先从左向右还是先从上到下，这个都可以。
5. 举例推导dp数组：见上图。

### 2 01背包问题+滚动dp数组

还是上面用那张图说明。

![21b4b58e1ce6f28000fb9936e52e7f5.jpg](https://www.misaka19327.cc/static/img/9f1a70f38b7b1287cf811e938ebafa87.21b4b58e1ce6f28000fb9936e52e7f5.jpg)

仅对于背包容量为3，取到第二个物品时那一种情况来说，**对它有意义的只有它上方那一排的值，或者更具体的来说，是它上面那一排的左侧和正上方的值**。而且01背包只需要知道最后能取到的最大价值是多少，并不关心过程。因此出现了一种使用一维数组求解01背包的写法，具体如下图：

![cec66a04e7cffe7ea6e1f750d59db57.jpg](https://www.misaka19327.cc/static/img/a10dfd43685201007e20fb2213909b08.cec66a04e7cffe7ea6e1f750d59db57.jpg)

一维数组的写法，就是去掉了二维数组中，已经求过并且已经没有任何的递推情况会使用的值之后的写法。这个写法就是抽象版本的二维数组，所以五部曲中大部分都可以直接用，只有遍历顺序需要注意，从上往下不变，但是**左右的遍历顺序必须是从右向左**，因为从左向右遍历的时候，按递推公式，它会直接取更新后的这一行的左侧，而不是还未更新的上一行的左侧。

### 3 LeetCode416 分割等和子集

:::info{title="题目"}
给你一个 只包含正整数 的 非空 数组 nums 。请你判断是否可以将这个数组分割成两个子集，使得两个子集的元素和相等。

 

示例 1：
```
输入：nums = [1,5,11,5]
输出：true
解释：数组可以分割成 [1, 5, 5] 和 [11] 。
```
示例 2：
```
输入：nums = [1,2,3,5]
输出：false
解释：数组不能分割成两个元素和相等的子集。
```

提示：

- 1 <= nums.length <= 200
- 1 <= nums[i] <= 100
:::

~~个人觉得，背包问题最难的不是背包问题本身，而是怎么把背包问题和现实问题对应上。~~

既然已经限定了是两个子集，而且目标是两个子集元素和相等，并且还允许不相等的情况，那目标就很明确了：在给定的数字中，挑出一些数字，使得它们的和为所有元素和的一半，找得到就返回true，找不到就返回false；如果所有元素和是个奇数，直接返回false就行。

之后再把这个问题和背包问题进行对应：子集就是背包本身，目标就是要求特定总和的子集，所以子集里面元素的和不能超过子集，而且确实要实时统计子集的和，所以每个数字都是一个物品，而且**这个物品的重量和价值都等于这个数字，这样才能保证背包能够装下（子集里面元素的和不能超过子集）而且实时看到背包能装的物品的最大价值（实时统计子集的和）**。之后就是按照01背包写就可以了，注意下例是一维dp数组的写法。

``` go
func canPartition(nums []int) bool {
    target := 0
    i := 0
    for {
        if i >= len(nums) {
            break
        }
        target += nums[i]
        i++
    }
    if target%2 != 0 {
        return false
    }
    target = target / 2
    dpArray := make([]int, target+1)
    for i := range nums { // 遍历物品
        for j := target; j >= nums[i]; j-- { // 遍历背包 这个循环条件是先确保物品能够放到这个背包里
            dpArray[j] = max(dpArray[j], dpArray[j - nums[i]] + nums[i])
        }
    }
    return dpArray[target] == target
}
```