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, (size63)/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等分布式缓存可以构建高效的多级缓存架构。在实际项目中需要根据业务需求选择合适的缓存策略处理好缓存一致性和缓存击穿等问题以构建稳定、高效的缓存系统。