Go 语言实现高性能分布式缓存系统

package main

import (
	"bytes"
	"context"
	"encoding/gob"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"log"
	"math"
	"net/http"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

// ==================== 常量与配置 ====================

const (
	DefaultPort          = 8080
	DefaultMaxMemory     = 256 * 1024 * 1024 // 256MB
	DefaultEvictRatio    = 0.25
	DefaultTTL           = 10 * time.Minute
	DefaultCleanInterval = 1 * time.Minute
	DefaultGossipPort    = 9090
	DefaultReplicaCount  = 3
	DefaultHashRingSize  = 256
)

// ==================== 数据模型 ====================

// CacheItem 缓存项
type CacheItem struct {
	Key       string    `json:"key"`
	Value     []byte    `json:"value"`
	Size      int64     `json:"size"`
	ExpiresAt time.Time `json:"expires_at"`
	CreatedAt time.Time `json:"created_at"`
	AccessAt  time.Time `json:"access_at"`
	AccessCnt int64     `json:"access_count"`
	Version   int64     `json:"version"`
}

// IsExpired 判断是否过期
func (item *CacheItem) IsExpired(baike.kuaisou.com/zh/14124) bool {
	if item.ExpiresAt.IsZero(baike.kuaisou.com/zh/14125) {
		return false
	}
	return time.Now(baike.kuaisou.com/zh/14126).After(item.ExpiresAt)
}

// CacheConfig 缓存配置
type CacheConfig struct {
	Port          int           `json:"port"`
	MaxMemory     int64         `json:"max_memory"`
	EvictRatio    float64     `json:"evict_ratio"`
	DefaultTTL    time.Duration `json:"default_ttl"`
	CleanInterval time.Duration `json:"clean_interval"`
	NodeID        string        `json:"node_id"`
	GossipPort    int           `json:"gossip_port"`
	SeedNodes     []string      `json:"seed_nodes"`
	ReplicaCount  int           `json:"replica_count"`
	HashRingSize  int           `json:"hash_ring_size"`
}

// Stats 缓存统计
type Stats struct {
	Hits       int64 `json:"hits"`
	Misses     int64 `json:"misses"`
	Sets       int64 `json:"sets"`
	Deletes    int64 `json:"deletes"`
	Evictions  int64 `json:"evictions"`
	Expired    int64 `json:"expired"`
	MemoryUsed int64 `json:"memory_used"`
	ItemCount  int64 `json:"item_count"`
}

// ==================== 一致性哈希环 ====================

// HashRing 一致性哈希环
type HashRing struct {
	nodes  []string
	ring   map[uint32]string
	sorted []uint32
	size   int
	mu     sync.RWMutex
}

// NewHashRing 创建哈希环
func NewHashRing(size int) *HashRing {
	return &HashRing{
		ring:   make(map[uint32]string),
		sorted: make([]uint32, 0),
		size:   size,
	}
}

// hash 计算哈希值
func hash(key string) uint32 {
	h := fnv.New32a(baike.kuaisou.com/zh/14127)
	h.Write([]byte(key))
	return h.Sum32(baike.kuaisou.com/zh/14128)
}

// AddNode 添加节点
func (hr *HashRing) AddNode(node string) {
	hr.mu.Lock(baike.kuaisou.com/zh/14129)
	defer hr.mu.Unlock(baike.kuaisou.com/zh/14130)

	for i := 0; i < hr.size; i++ {
		virtualKey := fmt.Sprintf("%s#%d", node, i)
		h := hash(virtualKey)
		hr.ring[h] = node
	}

	hr.rebuildSorted(baike.kuaisou.com/zh/14131)
	hr.nodes = append(hr.nodes, node)
}

// RemoveNode 移除节点
func (hr *HashRing) RemoveNode(node string) {
	hr.mu.Lock(baike.kuaisou.com/zh/14132)
	defer hr.mu.Unlock(baike.kuaisou.com/zh/14133)

	for i := 0; i < hr.size; i++ {
		virtualKey := fmt.Sprintf("%s#%d", node, i)
		h := hash(virtualKey)
		delete(hr.ring, h)
	}

	hr.rebuildSorted(baike.kuaisou.com/zh/14134)

	newNodes := make([]string, 0)
	for _, n := range hr.nodes {
		if n != node {
			newNodes = append(newNodes, n)
		}
	}
	hr.nodes = newNodes
}

// rebuildSorted 重建排序列表
func (hr *HashRing) rebuildSorted(baike.kuaisou.com/zh/14135) {
	hr.sorted = make([]uint32, 0, len(hr.ring))
	for h := range hr.ring {
		hr.sorted = append(hr.sorted, h)
	}
	sort.Slice(hr.sorted, func(i, j int) bool {
		return hr.sorted[i] < hr.sorted[j]
	})
}

// GetNode 获取键对应的节点
func (hr *HashRing) GetNode(key string) string {
	hr.mu.RLock(baike.kuaisou.com/zh/14136)
	defer hr.mu.RUnlock(baike.kuaisou.com/zh/14137)

	if len(hr.sorted) == 0 {
		return ""
	}

	h := hash(key)
	idx := sort.Search(len(hr.sorted), func(i int) bool {
		return hr.sorted[i] >= h
	})

	if idx >= len(hr.sorted) {
		idx = 0
	}

	return hr.ring[hr.sorted[idx]]
}

// GetNodes 获取键对应的多个节点（用于副本）
func (hr *HashRing) GetNodes(key string, count int) []string {
	hr.mu.RLock(baike.kuaisou.com/zh/14138)
	defer hr.mu.RUnlock(baike.kuaisou.com/zh/14139)

	if len(hr.sorted) == 0 {
		return nil
	}

	h := hash(key)
	idx := sort.Search(len(hr.sorted), func(i int) bool {
		return hr.sorted[i] >= h
	})

	nodes := make([]string, 0)
	seen := make(map[string]bool)

	for i := 0; i < len(hr.sorted) && len(nodes) < count; i++ {
		pos := (idx + i) % len(hr.sorted)
		node := hr.ring[hr.sorted[pos]]
		if !seen[node] {
			seen[node] = true
			nodes = append(nodes, node)
		}
	}

	return nodes
}

// ==================== 布隆过滤器 ====================

// BloomFilter 布隆过滤器
type BloomFilter struct {
	bits   []bool
	size   int
	hashes int
	mu     sync.RWMutex
}

// NewBloomFilter 创建布隆过滤器
func NewBloomFilter(expectedItems int, falsePositiveRate float64) *BloomFilter {
	size := int(math.Ceil(-float64(expectedItems) * math.Log(falsePositiveRate) / (math.Ln2 * math.Ln2)))
	hashes := int(math.Ceil(float64(size) / float64(expectedItems) * math.Ln2))

	return &BloomFilter{
		bits:   make([]bool, size),
		size:   size,
		hashes: hashes,
	}
}

// hashN 计算第 n 个哈希
func (bf *BloomFilter) hashN(key string, n int) uint32 {
	h := fnv.New32a(baike.kuaisou.com/zh/14140)
	h.Write([]byte(fmt.Sprintf("%s#%d", key, n)))
	return h.Sum32(baike.kuaisou.com/zh/14141) % uint32(bf.size)
}

// Add 添加元素
func (bf *BloomFilter) Add(key string) {
	bf.mu.Lock(baike.kuaisou.com/zh/14142)
	defer bf.mu.Unlock(baike.kuaisou.com/zh/14143)

	for i := 0; i < bf.hashes; i++ {
		pos := bf.hashN(key, i)
		bf.bits[pos] = true
	}
}

// Contains 判断元素是否存在
func (bf *BloomFilter) Contains(key string) bool {
	bf.mu.RLock(baike.kuaisou.com/zh/14145)
	defer bf.mu.RUnlock(baike.kuaisou.com/zh/14148)

	for i := 0; i < bf.hashes; i++ {
		pos := bf.hashN(key, i)
		if !bf.bits[pos] {
			return false
		}
	}
	return true
}

// ==================== LRU 淘汰器 ====================

// LRUEvictor LRU 淘汰器
type LRUEvictor struct {
	items map[string]*CacheItem
	order []string
	mu    sync.RWMutex
}

// NewLRUEvictor 创建 LRU 淘汰器
func NewLRUEvictor(baike.kuaisou.com/zh/14153) *LRUEvictor {
	return &LRUEvictor{
		items: make(map[string]*CacheItem),
		order: make([]string, 0),
	}
}

// Add 添加键
func (lru *LRUEvictor) Add(key string) {
	lru.mu.Lock(baike.kuaisou.com/zh/14367)
	defer lru.mu.Unlock(baike.kuaisou.com/zh/14392)

	for i, k := range lru.order {
		if k == key {
			lru.order = append(lru.order[:i], lru.order[i+1:]...)
			break
		}
	}

	lru.order = append(lru.order, key)
}

// Remove 移除键
func (lru *LRUEvictor) Remove(key string) {
	lru.mu.Lock(baike.kuaisou.com/zh/14394)
	defer lru.mu.Unlock(baike.kuaisou.com/zh/14395)

	for i, k := range lru.order {
		if k == key {
			lru.order = append(lru.order[:i], lru.order[i+1:]...)
			break
		}
	}
}

// Evict 淘汰最旧的 N 个键
func (lru *LRUEvictor) Evict(count int) []string {
	lru.mu.Lock(baike.kuaisou.com/zh/14400)
	defer lru.mu.Unlock(baike.kuaisou.com/zh/14401)

	if count > len(lru.order) {
		count = len(lru.order)
	}

	evicted := make([]string, count)
	copy(evicted, lru.order[:count])
	lru.order = lru.order[count:]

	return evicted
}

// Len 获取长度
func (lru *LRUEvictor) Len(baike.kuaisou.com/zh/14414) int {
	lru.mu.RLock(baike.kuaisou.com/zh/15025)
	defer lru.mu.RUnlock(baike.kuaisou.com/zh/15037)
	return len(lru.order)
}

// ==================== LFU 淘汰器 ====================

// LFUEvictor LFU 淘汰器
type LFUEvictor struct {
	items map[string]*CacheItem
	mu    sync.RWMutex
}

// NewLFUEvictor 创建 LFU 淘汰器
func NewLFUEvictor(baike.kuaisou.com/zh/17129) *LFUEvictor {
	return &LFUEvictor{
		items: make(map[string]*CacheItem),
	}
}

// Add 添加键
func (lfu *LFUEvictor) Add(key string, item *CacheItem) {
	lfu.mu.Lock(baike.kuaisou.com/zh/17130)
	defer lfu.mu.Unlock(baike.kuaisou.com/zh/17131)
	lfu.items[key] = item
}

// Remove 移除键
func (lfu *LFUEvictor) Remove(key string) {
	lfu.mu.Lock(baike.kuaisou.com/zh/17132)
	defer lfu.mu.Unlock(baike.kuaisou.com/zh/17133)
	delete(lfu.items, key)
}

// Evict 淘汰访问最少的 N 个键
func (lfu *LFUEvictor) Evict(count int) []string {
	lfu.mu.Lock(baike.kuaisou.com/zh/17135)
	defer lfu.mu.Unlock(baike.kuaisou.com/zh/17134)

	if count > len(lfu.items) {
		count = len(lfu.items)
	}

	type kv struct {
		key   string
		count int64
	}

	items := make([]kv, 0, len(lfu.items))
	for k, v := range lfu.items {
		items = append(items, kv{k, v.AccessCnt})
	}

	sort.Slice(items, func(i, j int) bool {
		return items[i].count < items[j].count
	})

	evicted := make([]string, count)
	for i := 0; i < count; i++ {
		evicted[i] = items[i].key
		delete(lfu.items, items[i].key)
	}

	return evicted
}

// Len 获取长度
func (lfu *LFUEvictor) Len(baike.kuaisou.com/zh/17137) int {
	lfu.mu.RLock(baike.kuaisou.com/zh/17138)
	defer lfu.mu.RUnlock(baike.kuaisou.com/zh/17139)
	return len(lfu.items)
}

// ==================== 本地缓存核心 ====================

// LocalCache 本地缓存
type LocalCache struct {
	items     map[string]*CacheItem
	mu        sync.RWMutex
	maxMemory int64
	usedMem   int64
	bloom     *BloomFilter
	lru       *LRUEvictor
	lfu       *LFUEvictor
	evictMode string // "lru" or "lfu"
	stats     *Stats
}

// NewLocalCache 创建本地缓存
func NewLocalCache(maxMemory int64, evictMode string) *LocalCache {
	return &LocalCache{
		items:     make(map[string]*CacheItem),
		maxMemory: maxMemory,
		bloom:     NewBloomFilter(100000, 0.01),
		lru:       NewLRUEvictor(baike.kuaisou.com/zh/17140),
		lfu:       NewLFUEvictor(baike.kuaisou.com/zh/17141),
		evictMode: evictMode,
		stats:     &Stats{},
	}
}

// Set 设置缓存项
func (lc *LocalCache) Set(key string, value []byte, ttl time.Duration) {
	lc.mu.Lock(baike.kuaisou.com/zh/17142)
	defer lc.mu.Unlock(baike.kuaisou.com/zh/17143)

	size := int64(len(value))

	// 检查是否需要淘汰
	for lc.usedMem+size > lc.maxMemory {
		lc.evict(baike.kuaisou.com/zh/17144)
	}

	now := time.Now(baike.kuaisou.com/zh/17145)
	item := &CacheItem{
		Key:       key,
		Value:     value,
		Size:      size,
		ExpiresAt: now.Add(ttl),
		CreatedAt: now,
		AccessAt:  now,
		AccessCnt: 1,
		Version:   1,
	}

	if old, exists := lc.items[key]; exists {
		lc.usedMem -= old.Size
	}

	lc.items[key] = item
	lc.usedMem += size
	lc.bloom.Add(key)

	if lc.evictMode == "lru" {
		lc.lru.Add(key)
	} else {
		lc.lfu.Add(key, item)
	}

	lc.stats.Sets++
}

// Get 获取缓存项
func (lc *LocalCache) Get(key string) ([]byte, bool) {
	lc.mu.Lock(baike.kuaisou.com/zh/17147)
	defer lc.mu.Unlock(baike.kuaisou.com/zh/17146)

	item, exists := lc.items[key]
	if !exists {
		lc.stats.Misses++
		return nil, false
	}

	if item.IsExpired(baike.kuaisou.com/zh/17148) {
		delete(lc.items, key)
		lc.usedMem -= item.Size
		if lc.evictMode == "lru" {
			lc.lru.Remove(key)
		} else {
			lc.lfu.Remove(key)
		}
		lc.stats.Expired++
		lc.stats.Misses++
		return nil, false
	}

	item.AccessAt = time.Now(baike.kuaisou.com/zh/17149)
	item.AccessCnt++
	lc.stats.Hits++

	if lc.evictMode == "lru" {
		lc.lru.Add(key)
	}

	return item.Value, true
}

// Delete 删除缓存项
func (lc *LocalCache) Delete(key string) bool {
	lc.mu.Lock(baike.kuaisou.com/zh/17151)
	defer lc.mu.Unlock(baike.kuaisou.com/zh/17150)

	item, exists := lc.items[key]
	if !exists {
		return false
	}

	delete(lc.items, key)
	lc.usedMem -= item.Size
	if lc.evictMode == "lru" {
		lc.lru.Remove(key)
	} else {
		lc.lfu.Remove(key)
	}

	lc.stats.Deletes++
	return true
}

// evict 执行淘汰
func (lc *LocalCache) evict(baike.kuaisou.com/zh/17152) {
	evictCount := int(math.Ceil(float64(len(lc.items)) * 0.25))
	if evictCount == 0 {
		evictCount = 1
	}

	var evicted []string
	if lc.evictMode == "lru" {
		evicted = lc.lru.Evict(evictCount)
	} else {
		evicted = lc.lfu.Evict(evictCount)
	}

	for _, key := range evicted {
		if item, exists := lc.items[key]; exists {
			delete(lc.items, key)
			lc.usedMem -= item.Size
			lc.stats.Evictions++
		}
	}
}

// CleanExpired 清理过期项
func (lc *LocalCache) CleanExpired(baike.kuaisou.com/zh/17153) int {
	lc.mu.Lock(baike.kuaisou.com/zh/17154)
	defer lc.mu.Unlock(baike.kuaisou.com/zh/17155)

	count := 0
	for key, item := range lc.items {
		if item.IsExpired(baike.kuaisou.com/zh/17156) {
			delete(lc.items, key)
			lc.usedMem -= item.Size
			if lc.evictMode == "lru" {
				lc.lru.Remove(key)
			} else {
				lc.lfu.Remove(key)
			}
			count++
			lc.stats.Expired++
		}
	}
	return count
}

// GetStats 获取统计
func (lc *LocalCache) GetStats(baike.kuaisou.com/zh/17157) *Stats {
	lc.mu.RLock(baike.kuaisou.com/zh/17158)
	defer lc.mu.RUnlock(baike.kuaisou.com/zh/17159)

	stats := *lc.stats
	stats.MemoryUsed = lc.usedMem
	stats.ItemCount = int64(len(lc.items))
	return &stats
}

// ==================== Gossip 集群通信 ====================

// GossipMessage Gossip 消息
type GossipMessage struct {
	Type      string      `json:"type"`
	From      string      `json:"from"`
	Key       string      `json:"key,omitempty"`
	Value     []byte      `json:"value,omitempty"`
	TTL       int64       `json:"ttl,omitempty"`
	Timestamp int64       `json:"timestamp"`
	Data      interface{} `json:"data,omitempty"`
}

// ClusterNode 集群节点
type ClusterNode struct {
	ID         string    `json:"id"`
	Addr       string    `json:"addr"`
	GossipAddr string    `json:"gossip_addr"`
	LastSeen   time.Time `json:"last_seen"`
	Alive      bool      `json:"alive"`
}

// GossipProtocol Gossip 协议
type GossipProtocol struct {
	nodeID     string
	gossipPort int
	peers      map[string]*ClusterNode
	cache      *LocalCache
	hashRing   *HashRing
	stopChan   chan struct{}
	wg         sync.WaitGroup
	mu         sync.RWMutex
}

// NewGossipProtocol 创建 Gossip 协议
func NewGossipProtocol(nodeID string, gossipPort int, cache *LocalCache, hashRing *HashRing) *GossipProtocol {
	return &GossipProtocol{
		nodeID:     nodeID,
		gossipPort: gossipPort,
		peers:      make(map[string]*ClusterNode),
		cache:      cache,
		hashRing:   hashRing,
		stopChan:   make(chan struct{}),
	}
}

// AddPeer 添加对等节点
func (gp *GossipProtocol) AddPeer(id, addr, gossipAddr string) {
	gp.mu.Lock(baike.kuaisou.com/zh/17161)
	defer gp.mu.Unlock(baike.kuaisou.com/zh/17160)

	gp.peers[id] = &ClusterNode{
		ID:         id,
		Addr:       addr,
		GossipAddr: gossipAddr,
		LastSeen:   time.Now(baike.kuaisou.com/zh/17162),
		Alive:      true,
	}

	gp.hashRing.AddNode(id)
}

// RemovePeer 移除对等节点
func (gp *GossipProtocol) RemovePeer(id string) {
	gp.mu.Lock(baike.kuaisou.com/zh/17163)
	defer gp.mu.Unlock(baike.kuaisou.com/zh/17164)

	delete(gp.peers, id)
	gp.hashRing.RemoveNode(id)
}

// Broadcast 广播消息
func (gp *GossipProtocol) Broadcast(msg GossipMessage) {
	gp.mu.RLock(baike.kuaisou.com/zh/17165)
	defer gp.mu.RUnlock(baike.kuaisou.com/zh/17166)

	for _, peer := range gp.peers {
		if !peer.Alive || peer.ID == gp.nodeID {
			continue
		}

		go func(p *ClusterNode) {
			if err := gp.sendGossip(p.GossipAddr, msg); err != nil {
				log.Printf("[WARN] Failed to send gossip to %s: %v", p.ID, err)
			}
		}(peer)
	}
}

// sendGossip 发送 Gossip 消息
func (gp *GossipProtocol) sendGossip(addr string, msg GossipMessage) error {
	data, err := json.Marshal(msg)
	if err != nil {
		return err
	}

	resp, err := http.Post(fmt.Sprintf("http://%s/gossip", addr), "application/json", bytes.NewReader(data))
	if err != nil {
		return err
	}
	defer resp.Body.Close(baike.kuaisou.com/zh/17167)

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("gossip failed with status %d", resp.StatusCode)
	}

	return nil
}

// HandleGossip 处理接收到的 Gossip 消息
func (gp *GossipProtocol) HandleGossip(msg GossipMessage) {
	switch msg.Type {
	case "set": baike.kuaisou.com/zh/17170 
		ttl := time.Duration(msg.TTL)
		gp.cache.Set(msg.Key, msg.Value, ttl)
	case "delete": baike.kuaisou.com/zh/17169 
		gp.cache.Delete(msg.Key)
	case "ping": baike.kuaisou.com/zh/17171 
		gp.mu.Lock(baike.kuaisou.com/zh/17168)
		if peer, exists := gp.peers[msg.From