7天用Go从零实现分布式缓存GeeCache（学习）（5）
Go 语言锁机制与 Geecache 项目中的加锁实现详解
一、Go 语言中的锁机制
Go 语言的 sync 包提供了多种用于并发控制的锁机制，以下是常用的几种：
1. sync.Mutex（互斥锁）
- 用途：控制对共享资源的独占访问，只允许一个 goroutine 持有锁，防止数据竞争（race condition）。
- 使用方法：调用
Lock()加锁，Unlock()解锁。 - 适用场景：适用于需要完全互斥的场景，例如对共享变量的写操作。
示例：
var mu sync.Mutex
func increment() {
mu.Lock()
counter++
mu.Unlock()
}
2. sync.RWMutex（读写锁）
- 用途：允许多个 goroutine 同时读取，但写操作是独占的，能提高读多写少场景的性能。
- 使用方法：
RLock()获取读锁，RUnlock()释放读锁；Lock()获取写锁，Unlock()释放写锁。
- 适用场景：适用于读多写少的场景，例如缓存、配置文件等。
示例：
var mu sync.RWMutex
func read() {
mu.RLock()
defer mu.RUnlock()
fmt.Println(counter)
}
func write() {
mu.Lock()
defer mu.Unlock()
counter++
}
3. sync.Once（单次锁）
- 用途：确保某些初始化操作只执行一次。
- 使用方法：调用
Do(func)，传入的函数只会执行一次，无论多少 goroutine 调用Do。 - 适用场景：适用于单次初始化的场景，例如单例模式或仅初始化一次的资源。
示例：
var once sync.Once
func initialize() {
once.Do(func() {
fmt.Println("Initializing...")
})
}
4. sync.Cond（条件变量）
- 用途：用于协调多个 goroutine 的等待和通知操作，通常配合
Mutex使用。 - 使用方法：
Wait()等待条件满足并自动释放锁；Signal()唤醒一个等待的 goroutine；Broadcast()唤醒所有等待的 goroutine。
- 适用场景：适用于需要等待条件的场景，例如生产者-消费者模型。
示例：
var mu sync.Mutex
var cond = sync.NewCond(&mu)
func waitForCondition() {
cond.L.Lock()
cond.Wait() // 等待通知
fmt.Println("Condition met")
cond.L.Unlock()
}
func signalCondition() {
cond.L.Lock()
cond.Signal() // 通知一个等待中的 goroutine
cond.L.Unlock()
}
5. sync.Map（并发安全的 Map）
- 用途：实现并发安全的键值对存储，适合高并发环境下的读写操作。
- 使用方法：提供
Store、Load、Delete、Range等方法来操作 map。 - 适用场景：适用于大量的读写操作，例如缓存。
示例：
var m sync.Map
func main() {
m.Store("key", "value")
if v, ok := m.Load("key"); ok {
fmt.Println(v)
}
m.Delete("key")
}
6. sync.WaitGroup（等待组）
- 用途：用于等待一组并发操作完成。
- 使用方法：
Add(n)添加要等待的 goroutine 数量；Done()表示某个 goroutine 完成；Wait()阻塞直到所有 goroutine 完成。
- 适用场景：适用于需要等待多个 goroutine 完成的场景，例如并发任务的汇总。
示例：
var wg sync.WaitGroup
func worker() {
defer wg.Done()
fmt.Println("Working...")
}
func main() {
wg.Add(2)
go worker()
go worker()
wg.Wait() // 等待所有 worker 完成
}
7. atomic 包（原子操作）
- 用途：提供底层的原子操作，避免使用锁，但依赖于硬件的原子指令。
- 使用方法：
atomic.AddInt32、atomic.LoadInt32、atomic.StoreInt32等。 - 适用场景：适合简单计数、状态切换等轻量级并发场景，避免锁开销。
示例：
import "sync/atomic"
var counter int32
func increment() {
atomic.AddInt32(&counter, 1)
}
二、Geecache 项目中的结构体组合与加锁实现
Geecache 是一个分布式缓存系统，类似于 Go 语言中的 Groupcache。它采用了多种并发控制机制来确保在高并发环境下的性能和数据一致性。以下将详细介绍各个文件中的结构体组合情况、函数调用关系、函数签名，以及代码中加锁的具体实现。
1. 项目目录结构
├── geecache
│ ├── byteview.go
│ ├── cache.go
│ ├── consistenthash
│ │ ├── consistenthash.go
│ │ └── consistenthash_test.go
│ ├── geecache.go
│ ├── go.mod
│ ├── http.go
│ ├── lru
│ │ └── lru.go
│ ├── peers.go
│ └── singleflight
│ └── singleflight.go
├── go.mod
└── main
└── main.go
2. 结构体组合情况
2.1 Group 结构体
文件：geecache/geecache.go
type Group struct {
name string // 缓存的命名空间
getter Getter // 缓存未命中时获取源数据的回调
mainCache cache // 并发安全的本地缓存
peers PeerPicker // 分布式场景下的节点选择器
loader *singleflight.Group // 防止缓存击穿的请求合并
}
- 组合关系：
- 包含
cache实例：Group内嵌了一个cache类型的字段mainCache，用于管理本地的缓存数据。 - 使用
Getter接口：Group通过getter字段持有一个实现了Getter接口的实例，用于在缓存未命中时加载源数据。 - 持有
PeerPicker接口：Group通过peers字段持有一个实现了PeerPicker接口的实例（如HTTPPool），用于在分布式环境中根据键选择远程节点。 - 使用
singleflight.Group：Group使用loader字段来持有一个singleflight.Group实例，防止并发请求相同的键时重复加载数据。
- 包含
2.2 cache 结构体
文件：geecache/cache.go
type cache struct {
mu sync.Mutex // 互斥锁，保护 lru 和 cacheBytes 的并发访问
lru *lru.Cache // LRU 缓存实例
cacheBytes int64 // 缓存的最大字节数
}
- 组合关系：
- 包含
lru.Cache实例：cache结构体内部持有一个指向lru.Cache的指针lru，用于实际存储缓存数据和执行 LRU 淘汰策略。 - 使用互斥锁：
cache使用互斥锁mu来保护对lru的并发访问，确保线程安全。
- 包含
2.3 lru.Cache 结构体
文件：geecache/lru/lru.go
type Cache struct {
maxBytes int64 // 最大缓存容量
nbytes int64 // 当前已使用的缓存容量
ll *list.List // 双向链表，记录访问顺序
cache map[string]*list.Element // 键到链表节点的映射
OnEvicted func(key string, value Value) // 可选的回调函数
}
- 组合关系：
- 使用标准库的
list.List：lru.Cache使用双向链表ll来记录缓存项的访问顺序，以便实现 LRU 淘汰策略。 - 使用字典存储缓存项：
cache字段是一个字典，映射键到链表节点，方便快速查找缓存项。 - 定义了
Value接口：缓存项的值需要实现Value接口，该接口定义了Len()方法，用于计算值的大小。
- 使用标准库的
2.4 ByteView 结构体
文件：geecache/byteview.go
type ByteView struct {
b []byte // 存储实际的数据
}
- 组合关系：
- 实现了
lru.Value接口：ByteView实现了Len()方法，满足lru.Cache对值类型的要求。 - 不可变性：
ByteView通过只暴露数据的拷贝，确保了数据的不可变性，防止外部修改缓存中的数据。
- 实现了
2.5 consistenthash.Map 结构体
文件：geecache/consistenthash/consistenthash.go
type Map struct {
hash Hash // 哈希函数
replicas int // 每个节点的虚拟节点数
keys []int // 哈希环，存储所有虚拟节点的哈希值
hashMap map[int]string // 虚拟节点与真实节点的映射
}
- 组合关系：
- 使用哈希函数：
Map通过hash字段持有一个哈希函数，用于计算键和节点的哈希值。 - 维护虚拟节点和真实节点的映射：
hashMap字段存储了虚拟节点的哈希值与真实节点名称的映射关系。
- 使用哈希函数：
2.6 HTTPPool 结构体
文件：geecache/http.go
type HTTPPool struct {
self string // 当前节点的地址
basePath string // HTTP 请求的基础路径
mu sync.Mutex // 互斥锁，保护 peers 和 httpGetters 的并发访问
peers *consistenthash.Map // 一致性哈希环
httpGetters map[string]*httpGetter // 远程节点的 HTTP 客户端映射
}
- 组合关系：
- 实现
PeerPicker接口：HTTPPool实现了PeerPicker接口中的PickPeer方法，能够根据键选择远程节点。 - 使用
consistenthash.Map：HTTPPool使用一致性哈希环peers来管理所有的节点，并根据键选择对应的节点。 - 维护节点到
httpGetter的映射：httpGetters字段存储了远程节点地址到httpGetter的映射，用于与远程节点进行通信。
- 实现
2.7 httpGetter 结构体
文件：geecache/http.go
type httpGetter struct {
baseURL string // 远程节点的地址
}
- 组合关系：
- 实现
PeerGetter接口：httpGetter实现了PeerGetter接口中的Get方法，能够通过 HTTP 从远程节点获取数据。
- 实现
2.8 PeerPicker 和 PeerGetter 接口
文件：geecache/peers.go
type PeerPicker interface {
PickPeer(key string) (peer PeerGetter, ok bool)
}
type PeerGetter interface {
Get(group string, key string) ([]byte, error)
}
- 组合关系：
Group使用PeerPicker：Group通过持有PeerPicker接口的实例，能够在分布式环境中根据键选择远程节点。HTTPPool实现PeerPicker：HTTPPool实现了PickPeer方法，能够根据键选择合适的远程节点。httpGetter实现PeerGetter：httpGetter实现了Get方法，能够从远程节点获取数据。
3. 函数调用关系概览
-
客户端请求数据：
- 调用
Group.Get(key)方法。- 尝试从本地缓存
mainCache获取数据。 - 如果缓存未命中，调用
Group.load(key)加载数据。- 使用
singleflight.Group.Do(key, fn)确保并发情况下只加载一次。 - 在
fn内部：- 如果配置了
peers，调用peers.PickPeer(key)选择远程节点。- 如果选择到远程节点，调用
PeerGetter.Get(group, key)从远程获取数据。
- 如果选择到远程节点，调用
- 如果没有远程节点或远程获取失败，调用
Group.getLocally(key)从本地数据源加载数据。- 调用
Getter.Get(key)从数据源获取数据。
- 调用
- 如果配置了
- 将获取到的数据通过
Group.populateCache(key, value)添加到本地缓存。
- 使用
- 尝试从本地缓存
- 将数据返回给客户端。
- 调用
-
HTTPPool 处理 HTTP 请求：
- 实现
http.Handler接口的ServeHTTP方法。- 解析请求路径，获取
group和key。 - 调用
Group.Get(key)获取数据。 - 将数据返回给客户端。
- 解析请求路径，获取
- 实现
-
httpGetter 作为远程节点的客户端：
- 实现
PeerGetter接口的Get方法。- 构造请求 URL。
- 发送 HTTP GET 请求，获取数据。
- 实现
4. 代码中加锁的详细说明
在 Geecache 项目中，为了确保在高并发环境下的线程安全，主要使用了 sync.Mutex 和 sync.WaitGroup。以下将详细说明各个文件中加锁的具体实现及其目的。
4.1 cache.go 中的加锁
文件：geecache/cache.go
type cache struct {
mu sync.Mutex // 互斥锁，保护 lru 和 cacheBytes 的并发访问
lru *lru.Cache // LRU 缓存实例
cacheBytes int64 // 缓存的最大字节数
}
- 加锁目的：保护
lru缓存和cacheBytes字段的并发访问，确保在多协程环境下对缓存的操作是安全的。
4.1.1 add 方法
func (c *cache) add(key string, value ByteView) {
c.mu.Lock() // 加锁，保护对 lru 的并发访问
defer c.mu.Unlock() // 函数退出时解锁
// 延迟初始化 lru 缓存
if c.lru == nil {
c.lru = lru.New(c.cacheBytes, nil)
}
c.lru.Add(key, value) // 向 lru 缓存中添加数据
}
- 加锁位置：方法开始时调用
c.mu.Lock()，方法结束时使用defer c.mu.Unlock()解锁。 - 加锁目的：
- 确保对
lru缓存的访问是线程安全的。 - 防止多个协程同时初始化
lru，导致竞态条件（race condition）。 - 保护
lru.Add的调用，因为lru.Cache本身不支持并发访问。
- 确保对
4.1.2 get 方法
func (c *cache) get(key string) (value ByteView, ok bool) {
c.mu.Lock() // 加锁，保护对 lru 的并发访问
defer c.mu.Unlock() // 函数退出时解锁
if c.lru == nil {
return
}
if v, ok := c.lru.Get(key); ok {
return v.(ByteView), ok
}
return
}
- 加锁位置：同样在方法开始时加锁，结束时解锁。
- 加锁目的：
- 保护对
lru缓存的读取操作。 - 防止并发情况下多个协程同时访问未初始化的
lru，导致空指针异常或其他竞态问题。 - 确保
lru.Get的操作是线程安全的。
- 保护对
4.2 singleflight.go 中的加锁
文件：geecache/singleflight/singleflight.go
type Group struct {
mu sync.Mutex // 互斥锁，保护 m 的并发访问
m map[string]*call // 存储正在进行的请求
}
- 加锁目的：保护
m字典的并发访问，防止多个协程同时对其进行读写操作。
4.2.1 Do 方法
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.m的访问，包括初始化和查找。 - 第一次解锁：如果发现已有相同的请求在进行中，立即解锁，等待已有请求完成。
- 第二次加锁：在请求完成后，再次加锁，删除
g.m中对应的key。 - 第二次解锁：删除后立即解锁。
- 第一次加锁：方法开始时，保护对
- 加锁目的：
- 第一次加锁：
- 确保对
g.m的初始化和查找是线程安全的。 - 防止多个协程同时对
g.m进行修改，导致竞态条件。
- 确保对
- 第二次加锁：
- 确保从
g.m中删除已完成的请求时，不会与其他协程发生冲突。
- 确保从
- 第一次加锁：
- 锁的粒度控制：
- 在持有锁期间，只执行必要的操作，尽快解锁，减少锁的持有时间，提高并发性能。
4.2.2 sync.WaitGroup 的使用
- 用途：使用
c.wg（sync.WaitGroup）让等待的协程阻塞，直到请求完成，避免重复请求。 - 与锁的配合：
- 锁用于保护对共享资源
g.m的访问。 WaitGroup用于协调请求的执行和等待，避免协程忙等或重复执行。
- 锁用于保护对共享资源
4.3 http.go 中的加锁
文件：geecache/http.go
type HTTPPool struct {
self string // 当前节点的地址
basePath string // HTTP 请求的基础路径
mu sync.Mutex // 互斥锁，保护 peers 和 httpGetters 的并发访问
peers *consistenthash.Map // 一致性哈希环
httpGetters map[string]*httpGetter // 远程节点的 HTTP 客户端映射
}
- 加锁目的：保护
peers和httpGetters的并发访问，确保节点列表的更新和读取是线程安全的。
4.3.1 Set 方法
func (p *HTTPPool) Set(peers ...string) {
p.mu.Lock()
defer p.mu.Unlock()
p.peers = consistenthash.New(defaultReplicas, nil)
p.peers.Add(peers...)
p.httpGetters = make(map[string]*httpGetter, len(peers))
for _, peer := range peers {
p.httpGetters[peer] = &httpGetter{baseURL: peer + p.basePath}
}
}
- 加锁位置：方法开始时加锁，方法结束时通过
defer解锁。 - 加锁目的：
- 确保对
peers和httpGetters的更新是原子性的，防止在更新过程中其他协程读取到不完整的数据。 - 防止多个协程同时调用
Set方法，导致数据竞争。
- 确保对
4.3.2 PickPeer 方法
func (p *HTTPPool) PickPeer(key string) (PeerGetter, bool) {
p.mu.Lock()
defer p.mu.Unlock()
if peer := p.peers.Get(key); peer != "" && peer != p.self {
p.Log("Pick peer %s", peer)
return p.httpGetters[peer], true
}
return nil, false
}
- 加锁位置：方法开始时加锁，方法结束时通过
defer解锁。 - 加锁目的：
- 保护对
peers和httpGetters的并发访问，确保读取到的节点信息是一致的。 - 防止在读取节点信息时，另一个协程正在修改节点列表，导致数据不一致或程序崩溃。
- 保护对
4.4 geecache.go 中的并发控制
文件：geecache/geecache.go
type Group struct {
name string // 缓存组的名称
getter Getter // 加载数据的回调接口
mainCache cache // 并发安全的本地缓存
peers PeerPicker // 节点选择器
loader *singleflight.Group // 防止缓存击穿的请求合并
}
- 并发控制方式：
- 对于本地缓存
mainCache，通过其内部的互斥锁mu进行保护。 - 对于防止缓存击穿，使用了
singleflight.Group，避免并发情况下的重复请求。 Group本身没有使用互斥锁，因为其内部的关键部分都已经由各自的组件进行了并发控制。
- 对于本地缓存
5. 加锁机制确保线程安全的方式
5.1 互斥锁 sync.Mutex
- 工作原理：互斥锁是一种用于保护共享资源的锁机制。在同一时刻，只有一个 goroutine 能够获得互斥锁，从而独占地访问被保护的资源。
- 在代码中的作用：
- 防止竞态条件：当多个 goroutine 同时读写共享资源时，可能会发生竞态条件，导致数据不一致或程序崩溃。通过互斥锁，可以确保共享资源的访问是互斥的，防止竞态发生。
- 保护临界区：临界区是指对共享资源进行访问的代码片段。在进入临界区之前加锁，退出时解锁，确保临界区内的代码不会被多个 goroutine 同时执行。
5.2 加锁的粒度控制
- 细粒度加锁：在代码中，尽量缩小锁的持有时间，只在需要保护的代码段内持有锁，其他时间尽快解锁，提高并发性能。
- 避免死锁：在加锁的过程中，注意锁的顺序，避免在多个锁之间形成循环等待，导致死锁。
三、生成的 Protobuf 代码与 Geecache 的集成
1. Protobuf 代码概述
您提供的 Protobuf 代码定义了两个消息类型：Request 和 Response，用于 Geecache 节点间的通信。
文件：geecachepb.proto（生成的 geecachepb 包）
// Code generated by protoc-gen-go. DO NOT EDIT.
// source: geecachepb.proto
package geecachepb
import (
fmt "fmt"
proto "github.com/golang/protobuf/proto"
math "math"
)
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
type Request struct {
Group string `protobuf:"bytes,1,opt,name=group,proto3" json:"group,omitempty"`
Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *Request) Reset() { *m = Request{} }
func (m *Request) String() string { return proto.CompactTextString(m) }
func (*Request) ProtoMessage() {}
func (*Request) Descriptor() ([]byte, []int) {
return fileDescriptor_889d0a4ad37a0d42, []int{0}
}
func (m *Request) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Request.Unmarshal(m, b)
}
func (m *Request) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_Request.Marshal(b, m, deterministic)
}
func (m *Request) XXX_Merge(src proto.Message) {
xxx_messageInfo_Request.Merge(m, src)
}
func (m *Request) XXX_Size() int {
return xxx_messageInfo_Request.Size(m)
}
func (m *Request) XXX_DiscardUnknown() {
xxx_messageInfo_Request.DiscardUnknown(m)
}
var xxx_messageInfo_Request proto.InternalMessageInfo
func (m *Request) GetGroup() string {
if m != nil {
return m.Group
}
return ""
}
func (m *Request) GetKey() string {
if m != nil {
return m.Key
}
return ""
}
type Response struct {
Value []byte `protobuf:"bytes,1,opt,name=value,proto3" json:"value,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *Response) Reset() { *m = Response{} }
func (m *Response) String() string { return proto.CompactTextString(m) }
func (*Response) ProtoMessage() {}
func (*Response) Descriptor() ([]byte, []int) {
return fileDescriptor_889d0a4ad37a0d42, []int{1}
}
func (m *Response) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Response.Unmarshal(m, b)
}
func (m *Response) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_Response.Marshal(b, m, deterministic)
}
func (m *Response) XXX_Merge(src proto.Message) {
xxx_messageInfo_Response.Merge(m, src)
}
func (m *Response) XXX_Size() int {
return xxx_messageInfo_Response.Size(m)
}
func (m *Response) XXX_DiscardUnknown() {
xxx_messageInfo_Response.DiscardUnknown(m)
}
var xxx_messageInfo_Response proto.InternalMessageInfo
func (m *Response) GetValue() []byte {
if m != nil {
return m.Value
}
return nil
}
func init() {
proto.RegisterType((*Request)(nil), "geecachepb.Request")
proto.RegisterType((*Response)(nil), "geecachepb.Response")
}
func init() { proto.RegisterFile("geecachepb.proto", fileDescriptor_889d0a4ad37a0d42) }
var fileDescriptor_889d0a4ad37a0d42 = []byte{
// 148 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0x48, 0x4f, 0x4d, 0x4d,
0x4e, 0x4c, 0xce, 0x48, 0x2d, 0x48, 0xd2, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x42, 0x88,
0x28, 0x19, 0x72, 0xb1, 0x07, 0xa5, 0x16, 0x96, 0xa6, 0x16, 0x97, 0x08, 0x89, 0x70, 0xb1, 0xa6,
0x17, 0xe5, 0x97, 0x16, 0x48, 0x30, 0x2a, 0x30, 0x6a, 0x70, 0x06, 0x41, 0x38, 0x42, 0x02, 0x5c,
0xcc, 0xd9, 0xa9, 0x95, 0x12, 0x4c, 0x60, 0x31, 0x10, 0x53, 0x49, 0x81, 0x8b, 0x23, 0x28, 0xb5,
0xb8, 0x20, 0x3f, 0xaf, 0x38, 0x15, 0xa4, 0xa7, 0x2c, 0x31, 0xa7, 0x34, 0x15, 0xac, 0x87, 0x27,
0x08, 0xc2, 0x31, 0xb2, 0xe3, 0xe2, 0x72, 0x07, 0x69, 0x76, 0x06, 0x59, 0x22, 0x64, 0xc0, 0xc5,
0xec, 0x9e, 0x5a, 0x22, 0x24, 0xac, 0x87, 0xe4, 0x10, 0xa8, 0x9d, 0x52, 0x22, 0xa8, 0x82, 0x10,
0x53, 0x93, 0xd8, 0xc0, 0xee, 0x34, 0x06, 0x04, 0x00, 0x00, 0xff, 0xff, 0x5c, 0xd5, 0xdd, 0x09,
0xbb, 0x00, 0x00, 0x00,
}
- 消息类型：
Request：包含Group和Key字段，用于客户端向缓存服务器请求数据。Response：包含Value字段，用于缓存服务器响应客户端的数据请求。
2. Protobuf 在 Geecache 中的作用
Protobuf（Protocol Buffers）是一种高效的序列化结构数据的工具，广泛用于服务间通信。在 Geecache 项目中，Protobuf 定义了节点间通信的消息格式，确保数据的结构化和高效传输。
-
定义：
Request：客户端向缓存节点发送的数据请求，包含要查询的Group和Key。Response：缓存节点返回的数据响应，包含对应的Value。
-
使用场景：
- 节点间通信：通过 HTTP 或其他协议传输 Protobuf 编码的数据，实现节点间的数据共享和同步。
- 数据序列化：确保数据在网络传输中的高效和一致性。
3. 集成 Protobuf 的 Geecache
在 Geecache 项目中，Protobuf 消息类型主要用于 HTTP 通信模块。具体来说，HTTPPool 负责处理 HTTP 请求，并使用 Protobuf 序列化和反序列化 Request 和 Response 消息，以实现高效的数据交换。
示例流程：
-
客户端发送请求：
- 客户端构造一个
Request消息，包含Group和Key。 - 使用 Protobuf 序列化
Request消息，并通过 HTTP 发送给目标缓存节点。
- 客户端构造一个
-
缓存节点处理请求：
- 接收到 HTTP 请求后，
HTTPPool的ServeHTTP方法解析请求，反序列化为Request消息。 - 调用
Group.Get(key)获取数据。 - 将数据封装在
Response消息中，使用 Protobuf 序列化后通过 HTTP 响应给客户端。
- 接收到 HTTP 请求后，
-
客户端接收响应：
- 客户端接收到 HTTP 响应，反序列化为
Response消息，获取Value数据。
- 客户端接收到 HTTP 响应，反序列化为
4. 完整的结构体交互图示
Client
|
v
HTTPPool (ServeHTTP)
|
v
Group.Get(key)
|
v
mainCache.get(key)
|
v
(lru.Cache.Get)
|
v
(if miss) -> loader.Do(key, fn)
|
v
PeerPicker.PickPeer(key)
|
+---------+---------+
| |
v v
PeerGetter.Get getLocally(key)
| |
v v
Response Getter.Get(key)
| |
v v
mainCache.add(key, value) |
| |
+---------+---------+
|
v
Return value
四、总结与建议
1. 并发控制的实现
在 Geecache 项目中，通过以下方式实现了高效且安全的并发控制：
-
互斥锁
sync.Mutex：cache结构体：使用mu锁保护对lru.Cache的读写操作，确保在高并发环境下缓存数据的一致性和安全性。singleflight.Group结构体：使用mu锁保护对正在进行的请求映射m的访问，防止重复请求和竞态条件。HTTPPool结构体：使用mu锁保护对一致性哈希环peers和 HTTP 客户端映射httpGetters的并发访问，确保节点选择和更新的线程安全。
-
等待组
sync.WaitGroup：singleflight.Group结构体：使用WaitGroup让等待的协程在数据加载完成后继续执行，避免重复加载相同的数据。
-
接口与组合：
PeerPicker和PeerGetter接口：通过接口抽象，实现了节点选择和数据获取的松耦合，增强了系统的灵活性和可扩展性。
2. 性能优化建议
-
使用
sync.RWMutex：- 对于读多写少的场景，可以考虑将
sync.Mutex替换为sync.RWMutex，允许多个协程同时读取，提高读操作的并发性能。
示例：
type cache struct { mu sync.RWMutex // 使用读写锁 lru *lru.Cache cacheBytes int64 } func (c *cache) add(key string, value ByteView) { c.mu.Lock() defer c.mu.Unlock() // 添加操作 } func (c *cache) get(key string) (value ByteView, ok bool) { c.mu.RLock() defer c.mu.RUnlock() // 获取操作 } - 对于读多写少的场景，可以考虑将
-
细化锁的粒度：
- 尽量缩小锁的持有范围，只在必要的代码块内加锁，减少锁的竞争，提高并发性能。
-
利用
sync.Map：- 对于高度并发的场景，可以考虑使用
sync.Map替代传统的加锁 map，以减少锁的开销。
- 对于高度并发的场景，可以考虑使用
3. 错误处理与日志记录
-
增强错误处理：
- 在网络请求和数据加载过程中，增加详细的错误处理和日志记录，便于调试和维护。
-
日志优化：
- 统一日志格式和级别，确保在不同模块间的日志信息清晰且有用。
4. 测试与监控
-
编写单元测试：
- 为各个模块编写详细的单元测试，尤其是并发场景下的测试，确保代码的正确性和稳定性。
-
性能监控：
- 添加缓存命中率、请求次数、并发请求数等指标的监控，帮助识别性能瓶颈并进行优化。
5. 配置管理
- 参数化配置：
- 将硬编码的参数（如虚拟节点数量、HTTP 基础路径）抽取到配置文件或环境变量中，增强系统的灵活性和可配置性。