7天用Go从零实现分布式缓存GeeCache（学习）（5）

🕗 发布于 2024-11-13 09:00 golang 分布式缓存

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 响应，反序列化为 Response 消息，获取 Value 数据。

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 基础路径）抽取到配置文件或环境变量中，增强系统的灵活性和可配置性。

原文地址：https://blog.csdn.net/weixin_51147313/article/details/143718512

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：牛客周赛 Round 67
下一篇：前端模拟面试：7个JavaScript数组进阶面试题

C#调试项目_附加进程调试
c#项目附加进程一级目录二级目录三级目录一级目录二级目录三级目录
阅读更多2024-11-15
线程-2-线程概念与控制
线程概念与控制，局部性原理， POXIS接口，分页式存储结构，页表结构，缺页中断
阅读更多2024-11-15
洛谷 P4011 孤岛营救问题（BFS分层图最短路，状态压缩）
我们考虑对已获得的钥匙进行二进制状态压缩。注意：同一个格子可能有多把钥匙，一开始的。显然，我们直接使用BFS求最短路即可。这个格子，且当前已有钥匙的状态为。
阅读更多2024-11-15
聊天服务器(3)muduo网络库
muduo只能装在linux中，依赖boost库客户端并不需要高并发。
阅读更多2024-11-15
Unity学习---IL2CPP打包时可能遇到的问题
在这种情况下一般我们把Managed Stripping Level设置为High，然后看一下报错的提示，往link.wxl文件中添加要保留的东西即可。IL2CPP打包后会自动对Unity工程的dll
阅读更多2024-11-15
OpenGL C++视频中添加图片及文字水印播放并录制
根据前置储备，该功能的实现流程：创建一个GLThread的OpenGL运行环境；YUV数据绘制到OpenGL纹理中；图片纹理创建绘制到glViewport；文字渲染绘制到glViewport；录制视频
阅读更多2024-11-15
java实现中小企业的erp系统
java实现中小企业的erp系统
阅读更多2024-11-15
智能零售柜商品识别
本项目在智能零售商品识别的背景下，采用了多项技术创新和优化，旨在提高商品检测与识别的效率和准确性。首先，使用了PaddleX作为训练框架，这是一个高效、灵活的深度学习平台，简化了模型训练和部署的复杂性
阅读更多2024-11-15
H.265流媒体播放器EasyPlayer.js网页直播/点播播放器WebGL: CONTEXT_LOST_WEBGL错误引发的原因
EasyPlayer.js播放器不仅支持H.264与H.265视频编码格式，还具备直播、点播、录像、快照截图、MP4播放、多屏播放、倍数播放、全屏播放等功能特性，兼容Windows、Linux、And
阅读更多2024-11-15
ubuntu连接orangepi-zero-2w桌面的几种方法
线连接 Orange Pi 开发板和 HDMI 显示器。然后打开 Nomachine 图形化连接开发板。例如使用finalshell 连接开发板。其中视频采集卡是usb输出，hdmi输入。线 +
阅读更多2024-11-15

7天用Go从零实现分布式缓存GeeCache（学习）（5）

Go 语言锁机制与 Geecache 项目中的加锁实现详解

一、Go 语言中的锁机制

1. sync.Mutex（互斥锁）

2. sync.RWMutex（读写锁）

3. sync.Once（单次锁）

4. sync.Cond（条件变量）

5. sync.Map（并发安全的 Map）

6. sync.WaitGroup（等待组）

7. atomic 包（原子操作）

二、Geecache 项目中的结构体组合与加锁实现

1. 项目目录结构

2. 结构体组合情况

2.1 Group 结构体

2.2 cache 结构体

2.3 lru.Cache 结构体

2.4 ByteView 结构体

2.5 consistenthash.Map 结构体

2.6 HTTPPool 结构体

2.7 httpGetter 结构体

2.8 PeerPicker 和 PeerGetter 接口

3. 函数调用关系概览

4. 代码中加锁的详细说明

4.1 cache.go 中的加锁

4.1.1 add 方法

4.1.2 get 方法

4.2 singleflight.go 中的加锁

4.2.1 Do 方法

4.2.2 sync.WaitGroup 的使用

4.3 http.go 中的加锁

4.3.1 Set 方法

4.3.2 PickPeer 方法

4.4 geecache.go 中的并发控制

5. 加锁机制确保线程安全的方式

5.1 互斥锁 sync.Mutex

5.2 加锁的粒度控制

三、生成的 Protobuf 代码与 Geecache 的集成

1. Protobuf 代码概述

2. Protobuf 在 Geecache 中的作用

3. 集成 Protobuf 的 Geecache

4. 完整的结构体交互图示

四、总结与建议

1. 并发控制的实现

2. 性能优化建议

3. 错误处理与日志记录

4. 测试与监控

5. 配置管理

相关文章

1. `sync.Mutex`（互斥锁）

2. `sync.RWMutex`（读写锁）

3. `sync.Once`（单次锁）

4. `sync.Cond`（条件变量）

5. `sync.Map`（并发安全的 Map）

6. `sync.WaitGroup`（等待组）

7. `atomic` 包（原子操作）

2.1 `Group` 结构体

2.2 `cache` 结构体

2.3 `lru.Cache` 结构体

2.4 `ByteView` 结构体

2.5 `consistenthash.Map` 结构体

2.6 `HTTPPool` 结构体

2.7 `httpGetter` 结构体

2.8 `PeerPicker` 和 `PeerGetter` 接口

4.1 `cache.go` 中的加锁

4.1.1 `add` 方法

4.1.2 `get` 方法

4.2 `singleflight.go` 中的加锁

4.2.1 `Do` 方法

4.2.2 `sync.WaitGroup` 的使用

4.3 `http.go` 中的加锁

4.3.1 `Set` 方法

4.3.2 `PickPeer` 方法

4.4 `geecache.go` 中的并发控制

5.1 互斥锁 `sync.Mutex`