Go语言实现分布式缓存：从LRU到多级缓存架构
Go语言实现分布式缓存：从LRU到多级缓存架构
引言
分布式缓存是现代分布式系统中提升性能的关键组件。Go语言凭借其高性能和并发支持，成为实现分布式缓存的理想选择。本文将深入探讨Go语言实现分布式缓存的完整实践。
一、缓存基础
1.1 缓存架构
┌─────────────────────────────────────────────────────────────┐ │ 缓存架构设计 │ ├─────────────────────────────────────────────────────────────┤ │ Client │ Local Cache │ Distributed Cache │ Database│ │ │ (LRU/ARC) │ (Redis/Memcached)│ │ │ 请求 │──> 查找缓存 ──│──> 查找分布式缓存 ──│──> 查询DB │ │ │ │ │ │ │ │<── 返回数据 ──│<── 返回数据 ────│<── 返回数据│ └─────────────────────────────────────────────────────────────┘1.2 缓存策略对比
| 策略 | 说明 | 适用场景 |
|---|---|---|
| LRU | 最近最少使用 | 通用场景 |
| LFU | 最不经常使用 | 访问频率差异大 |
| ARC | 自适应替换 | 混合场景 |
| FIFO | 先进先出 | 简单场景 |
二、本地LRU缓存实现
2.1 基础LRU实现
package lru import ( "container/list" "sync" ) type Cache struct { mu sync.Mutex maxBytes int64 nbytes int64 ll *list.List cache map[string]*list.Element OnEvicted func(key string, value Value) } type entry struct { key string value Value } type Value interface { Len() int } func New(maxBytes int64, onEvicted func(string, Value)) *Cache { return &Cache{ maxBytes: maxBytes, ll: list.New(), cache: make(map[string]*list.Element), OnEvicted: onEvicted, } } func (c *Cache) Get(key string) (value Value, ok bool) { c.mu.Lock() defer c.mu.Unlock() if ele, ok := c.cache[key]; ok { c.ll.MoveToFront(ele) kv := ele.Value.(*entry) return kv.value, true } return } func (c *Cache) RemoveOldest() { c.mu.Lock() defer c.mu.Unlock() ele := c.ll.Back() if ele != nil { c.ll.Remove(ele) kv := ele.Value.(*entry) delete(c.cache, kv.key) c.nbytes -= int64(len(kv.key)) + int64(kv.value.Len()) if c.OnEvicted != nil { c.OnEvicted(kv.key, kv.value) } } } func (c *Cache) Add(key string, value Value) { c.mu.Lock() defer c.mu.Unlock() if ele, ok := c.cache[key]; ok { c.ll.MoveToFront(ele) kv := ele.Value.(*entry) c.nbytes += int64(value.Len()) - int64(kv.value.Len()) kv.value = value } else { ele := c.ll.PushFront(&entry{key, value}) c.cache[key] = ele c.nbytes += int64(len(key)) + int64(value.Len()) } for c.maxBytes != 0 && c.maxBytes < c.nbytes { c.RemoveOldest() } } func (c *Cache) Len() int { c.mu.Lock() defer c.mu.Unlock() return c.ll.Len() }2.2 并发安全优化
type ConcurrentCache struct { shards int caches []*Cache } func NewConcurrentCache(shards int, maxBytes int64) *ConcurrentCache { caches := make([]*Cache, shards) for i := 0; i < shards; i++ { caches[i] = New(maxBytes/int64(shards), nil) } return &ConcurrentCache{ shards: shards, caches: caches, } } func (c *ConcurrentCache) getShard(key string) *Cache { hash := fnv.New32a() hash.Write([]byte(key)) return c.caches[hash.Sum32()%uint32(c.shards)] } func (c *ConcurrentCache) Get(key string) (Value, bool) { return c.getShard(key).Get(key) } func (c *ConcurrentCache) Add(key string, value Value) { c.getShard(key).Add(key, value) }三、Redis客户端实现
3.1 基础客户端
package redis import ( "context" "time" "github.com/go-redis/redis/v8" ) type Client struct { client *redis.Client } func NewClient(addr, password string, db int) *Client { return &Client{ client: redis.NewClient(&redis.Options{ Addr: addr, Password: password, DB: db, }), } } func (c *Client) Get(ctx context.Context, key string) (string, error) { return c.client.Get(ctx, key).Result() } func (c *Client) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) error { return c.client.Set(ctx, key, value, expiration).Err() } func (c *Client) Delete(ctx context.Context, keys ...string) error { return c.client.Del(ctx, keys...).Err() }3.2 缓存策略封装
type Cache struct { client *Client local *ConcurrentCache localTTL time.Duration remoteTTL time.Duration } func NewCache(client *Client, localSize int64) *Cache { return &Cache{ client: client, local: NewConcurrentCache(16, localSize), localTTL: 5 * time.Minute, remoteTTL: 30 * time.Minute, } } func (c *Cache) Get(ctx context.Context, key string) (string, error) { if val, ok := c.local.Get(key); ok { return val.(string), nil } val, err := c.client.Get(ctx, key) if err != nil { return "", err } c.local.Add(key, StringValue(val)) return val, nil } func (c *Cache) Set(ctx context.Context, key string, value string) error { if err := c.client.Set(ctx, key, value, c.remoteTTL); err != nil { return err } c.local.Add(key, StringValue(value)) return nil } type StringValue string func (s StringValue) Len() int { return len(s) }四、多级缓存架构
4.1 架构设计
type MultiLevelCache struct { local *ConcurrentCache remote *redis.Client source DataSource } type DataSource interface { Get(ctx context.Context, key string) (string, error) } func NewMultiLevelCache(remote *redis.Client, source DataSource) *MultiLevelCache { return &MultiLevelCache{ local: NewConcurrentCache(16, 1024*1024*100), // 100MB remote: remote, source: source, } } func (m *MultiLevelCache) Get(ctx context.Context, key string) (string, error) { if val, ok := m.local.Get(key); ok { return val.(string), nil } val, err := m.remote.Get(ctx, key).Result() if err == nil { m.local.Add(key, StringValue(val)) return val, nil } if err != redis.Nil { return "", err } val, err = m.source.Get(ctx, key) if err != nil { return "", err } m.remote.Set(ctx, key, val, 30*time.Minute) m.local.Add(key, StringValue(val)) return val, nil }4.2 缓存预热
func (m *MultiLevelCache) Warmup(ctx context.Context, keys []string) error { pipeline := m.remote.Pipeline() for _, key := range keys { pipeline.Get(ctx, key) } results, err := pipeline.Exec(ctx) if err != nil { return err } for i, result := range results { if result != nil { val, _ := result.(*redis.StringCmd).Result() m.local.Add(keys[i], StringValue(val)) } } return nil }五、缓存一致性策略
5.1 写穿透
func (m *MultiLevelCache) WriteThrough(ctx context.Context, key string, value string) error { if err := m.source.Set(ctx, key, value); err != nil { return err } m.remote.Set(ctx, key, value, 30*time.Minute) m.local.Add(key, StringValue(value)) return nil }5.2 写回
type WriteBackCache struct { *MultiLevelCache dirty map[string]bool mu sync.Mutex } func NewWriteBackCache(remote *redis.Client, source DataSource) *WriteBackCache { return &WriteBackCache{ MultiLevelCache: NewMultiLevelCache(remote, source), dirty: make(map[string]bool), } } func (w *WriteBackCache) Set(ctx context.Context, key string, value string) error { w.mu.Lock() w.dirty[key] = true w.mu.Unlock() w.local.Add(key, StringValue(value)) return nil } func (w *WriteBackCache) Flush(ctx context.Context) error { w.mu.Lock() defer w.mu.Unlock() for key := range w.dirty { if val, ok := w.local.Get(key); ok { if err := w.source.Set(ctx, key, val.(string)); err != nil { return err } w.remote.Set(ctx, key, val, 30*time.Minute) } delete(w.dirty, key) } return nil }六、缓存击穿解决方案
6.1 互斥锁方案
type CacheWithLock struct { *MultiLevelCache locks sync.Map } func (c *CacheWithLock) Get(ctx context.Context, key string) (string, error) { if val, ok := c.local.Get(key); ok { return val.(string), nil } lockKey := "lock:" + key lock, _ := c.locks.LoadOrStore(lockKey, &sync.Mutex{}) mutex := lock.(*sync.Mutex) mutex.Lock() defer mutex.Unlock() if val, ok := c.local.Get(key); ok { return val.(string), nil } val, err := c.source.Get(ctx, key) if err != nil { return "", err } c.remote.Set(ctx, key, val, 30*time.Minute) c.local.Add(key, StringValue(val)) return val, nil }6.2 布隆过滤器
type BloomFilter struct { bitset []uint64 hashes int } func NewBloomFilter(size int, hashes int) *BloomFilter { return &BloomFilter{ bitset: make([]uint64, (size+63)/64), hashes: hashes, } } func (b *BloomFilter) Add(key string) { for i := 0; i < b.hashes; i++ { hash := b.hash(key, i) b.bitset[hash/64] |= 1 << (hash % 64) } } func (b *BloomFilter) Contains(key string) bool { for i := 0; i < b.hashes; i++ { hash := b.hash(key, i) if (b.bitset[hash/64] & (1 << (hash % 64))) == 0 { return false } } return true } func (b *BloomFilter) hash(key string, seed int) uint64 { h := fnv.New64a() h.Write([]byte(key)) return h.Sum64() ^ uint64(seed*1103515245) } func (c *CacheWithLock) GetWithBloom(ctx context.Context, key string, bf *BloomFilter) (string, error) { if !bf.Contains(key) { return "", ErrNotFound } return c.Get(ctx, key) }七、实战：分布式缓存服务
type CacheService struct { cache *MultiLevelCache server *http.Server } func NewCacheService(redisAddr string, source DataSource) *CacheService { client := redis.NewClient(&redis.Options{ Addr: redisAddr, }) return &CacheService{ cache: NewMultiLevelCache(client, source), } } func (s *CacheService) ServeHTTP(w http.ResponseWriter, r *http.Request) { key := r.URL.Query().Get("key") if key == "" { http.Error(w, "key is required", http.StatusBadRequest) return } val, err := s.cache.Get(r.Context(), key) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.Write([]byte(val)) } func (s *CacheService) Start(addr string) error { s.server = &http.Server{ Addr: addr, Handler: s, } return s.server.ListenAndServe() }结论
分布式缓存是提升系统性能的关键组件。通过结合本地LRU缓存和Redis等分布式缓存，可以构建高效的多级缓存架构。在实际项目中，需要根据业务需求选择合适的缓存策略，处理好缓存一致性和缓存击穿等问题，以构建稳定、高效的缓存系统。