🔥 缓存三大问题：机制与原理详解

package cache

import (
    "context"
    "math/rand"
    "time"

    "github.com/redis/go-redis/v9"
)

// RandomTTLCache 随机TTL缓存实现
type RandomTTLCache struct {
    rdb       *redis.Client
    baseTTL   time.Duration // 基础过期时间
    jitterMax time.Duration // 随机抖动最大值
}

func NewRandomTTLCache(rdb *redis.Client, baseTTL, jitterMax time.Duration) *RandomTTLCache {
    return &RandomTTLCache{
        rdb:       rdb,
        baseTTL:   baseTTL,
        jitterMax: jitterMax,
    }
}

// Set 设置缓存（带随机TTL）
func (c *RandomTTLCache) Set(ctx context.Context, key string, value interface{}) error {
    // 计算实际TTL = 基础TTL + 随机抖动
    actualTTL := c.baseTTL + time.Duration(rand.Int63n(int64(c.jitterMax)))
  
    return c.rdb.Set(ctx, key, value, actualTTL).Err()
}

// SetBatch 批量设置缓存（每个Key独立随机TTL）
func (c *RandomTTLCache) SetBatch(ctx context.Context, items map[string]interface{}) error {
    pipe := c.rdb.Pipeline()
  
    for key, value := range items {
        // 每个Key使用独立的随机TTL
        actualTTL := c.baseTTL + time.Duration(rand.Int63n(int64(c.jitterMax)))
        pipe.Set(ctx, key, value, actualTTL)
    }
  
    _, err := pipe.Exec(ctx)
    return err
}

// 使用示例
func Example() {
    rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
  
    // 基础TTL 30分钟，随机抖动 0~5分钟
    cache := NewRandomTTLCache(rdb, 30*time.Minute, 5*time.Minute)
  
    // 设置商品缓存
    ctx := context.Background()
    cache.Set(ctx, "product:1001", productData1) // 可能 32分钟后过期
    cache.Set(ctx, "product:1002", productData2) // 可能 34分钟后过期
    cache.Set(ctx, "product:1003", productData3) // 可能 31分钟后过期
    // ... 每个Key的过期时间都不同，避免雪崩
}

package cache

import (
    "context"
    "errors"
    "math/rand"
    "sync"
    "time"

    "github.com/redis/go-redis/v9"
    "golang.org/x/time/rate"
)

// AvalancheProtection 雪崩综合防护
type AvalancheProtection struct {
    rdb       *redis.Client
    limiter   *rate.Limiter      // 限流器
    semaphore chan struct{}     // 信号量，控制并发查DB的数量
    mu        sync.Mutex
    baseTTL   time.Duration
    jitterMax time.Duration
}

func NewAvalancheProtection(rdb *redis.Client, maxDBConcurrency int) *AvalancheProtection {
    return &AvalancheProtection{
        rdb:       rdb,
        limiter:   rate.NewLimiter(1000, 2000), // 每秒1000个请求，桶容量2000
        semaphore: make(chan struct{}, maxDBConcurrency), // 最多N个goroutine同时查DB
        baseTTL:   30 * time.Minute,
        jitterMax: 10 * time.Minute,
    }
}

// Get 带多级保护的缓存查询
func (p *AvalancheProtection) Get(
    ctx context.Context,
    key string,
    loadFn func(ctx context.Context) (interface{}, error),
) (interface{}, error) {
    // 第一层保护：限流
    if !p.limiter.Allow() {
        return nil, errors.New("rate limit exceeded, please retry later")
    }

    // 尝试从缓存获取
    val, err := p.rdb.Get(ctx, key).Result()
    if err == nil {
        return val, nil
    }

    // 第二层保护：信号量控制并发查DB数量
    select {
    case p.semaphore <- struct{}{}:
        defer func() { <-p.semaphore }()
    case <-time.After(3 * time.Second):
        // 等待超时，返回降级数据
        return p.getFallbackData(key)
    case <-ctx.Done():
        return nil, ctx.Err()
    }

    // 第三层保护：分布式锁，防止同一Key重复查DB
    lockKey := "lock:" + key
    locked := p.rdb.SetNX(ctx, lockKey, "1", 10*time.Second).Val()
    if !locked {
        // 其他goroutine正在重建该Key的缓存，等待
        time.Sleep(100 * time.Millisecond)
        // 重试从缓存获取
        return p.rdb.Get(ctx, key).Result()
    }
    defer p.rdb.Del(ctx, lockKey)

    // 查询数据库
    data, err := loadFn(ctx)
    if err != nil {
        return p.getFallbackData(key) // 查询失败，返回降级数据
    }

    // 写入缓存（随机TTL）
    actualTTL := p.baseTTL + time.Duration(rand.Int63n(int64(p.jitterMax)))
    p.rdb.Set(ctx, key, data, actualTTL)

    return data, nil
}

// getFallbackData 获取降级数据
func (p *AvalancheProtection) getFallbackData(key string) (interface{}, error) {
    // 降级策略：返回默认值/上次缓存的快照/空结果
    return map[string]interface{}{
        "status":  "degraded",
        "message": "service temporarily unavailable",
    }, nil
}

package cache

import (
    "context"
    "fmt"
    "time"

    "github.com/redis/go-redis/v9"
)

// NewRedisCluster 创建 Redis Cluster 客户端
func NewRedisCluster() *redis.ClusterClient {
    return redis.NewClusterClient(&redis.ClusterOptions{
        // 集群节点地址（至少配置一个，客户端会自动发现其他节点）
        Addrs: []string{
            "192.168.1.10:7000",
            "192.168.1.10:7001",
            "192.168.1.11:7000",
            "192.168.1.11:7001",
            "192.168.1.12:7000",
            "192.168.1.12:7001",
        },
      
        // 密码认证
        Password: "your_password",
      
        // 最大重试次数（故障转移期间自动重试）
        MaxRedirects: 3,
      
        // 读取超时
        ReadTimeout: 3 * time.Second,
      
        // 写入超时
        WriteTimeout: 3 * time.Second,
      
        // 连接池配置
        PoolSize:     100,
        MinIdleConns: 20,
      
        // 路由函数：读请求可以路由到从节点（读写分离）
        RouteByLatency: true,
      
        // 集群拓扑刷新间隔
        ClusterSlotsRefreshInterval: 10 * time.Second,
    })
}

// SentinelClient 哨兵模式客户端
func NewSentinelClient() *redis.Client {
    return redis.NewFailoverClient(&redis.FailoverOptions{
        // 哨兵节点地址
        SentinelAddrs: []string{
            "192.168.1.10:26379",
            "192.168.1.11:26379",
            "192.168.1.12:26379",
        },
      
        // 主节点名称（在哨兵配置中定义的）
        MasterName: "mymaster",
      
        // 数据库密码
        Password: "your_password",
      
        // 数据库编号
        DB: 0,
      
        // 只读节点（读写分离）
        ReplicaOnly: false,
    })
}

// HealthCheck 缓存健康检查
func HealthCheck(ctx context.Context, client *redis.ClusterClient) error {
    if err := client.Ping(ctx).Err(); err != nil {
        return fmt.Errorf("redis cluster health check failed: %w", err)
    }
  
    // 检查集群状态
    clusterInfo, err := client.ClusterInfo(ctx).Result()
    if err != nil {
        return fmt.Errorf("failed to get cluster info: %w", err)
    }
  
    // 验证集群状态是否为ok
    if !contains(clusterInfo, "cluster_state:ok") {
        return fmt.Errorf("cluster state is not ok")
    }
  
    return nil
}

package handler

import (
    "errors"
    "fmt"
    "regexp"
    "strconv"
)

var (
    ErrInvalidID     = errors.New("invalid parameter: id must be positive")
    ErrInvalidFormat = errors.New("invalid parameter format")
    ErrIDOutOfRange  = errors.New("id out of valid range")
)

// ParamValidator 参数校验器
type ParamValidator struct {
    maxUserID int64 // 最大合法用户ID（可根据业务设定）
}

func NewParamValidator(maxUserID int64) *ParamValidator {
    return &ParamValidator{maxUserID: maxUserID}
}

// ValidateUserID 校验用户ID
func (v *ParamValidator) ValidateUserID(idStr string) (int64, error) {
    // 1. 非空校验
    if idStr == "" {
        return 0, fmt.Errorf("%w: id is empty", ErrInvalidFormat)
    }

    // 2. 格式校验（纯数字）
    matched, _ := regexp.MatchString(`^\d+$`, idStr)
    if !matched {
        return 0, fmt.Errorf("%w: id must be numeric", ErrInvalidFormat)
    }

    // 3. 数值转换与范围校验
    id, err := strconv.ParseInt(idStr, 10, 64)
    if err != nil {
        return 0, fmt.Errorf("%w: %v", ErrInvalidFormat, err)
    }

    // 4. 正数校验
    if id <= 0 {
        return 0, ErrInvalidID
    }

    // 5. 范围校验（防止超大ID）
    if id > v.maxUserID {
        return 0, fmt.Errorf("%w: max is %d", ErrIDOutOfRange, v.maxUserID)
    }

    return id, nil
}

// GetUserHandler 用户查询接口
func (h *Handler) GetUserHandler(idStr string) (interface{}, error) {
    // 第一道防线：参数校验
    userID, err := h.validator.ValidateUserID(idStr)
    if err != nil {
        // 参数不合法，直接拒绝，不查缓存也不查数据库
        return nil, fmt.Errorf("bad request: %w", err)
    }

    // 参数合法，继续走缓存查询逻辑...
    return h.cache.Get(ctx, fmt.Sprintf("user:%d", userID), ...)
}

package cache

import (
    "context"
    "encoding/json"
    "errors"
    "time"

    "github.com/redis/go-redis/v9"
)

var (
    ErrDataNotFound = errors.New("data not found")
)

// 空值标记常量
const (
    NullValue     = "@@NULL@@"          // 空值标记字符串
    NullTTL       = 60 * time.Second    // 空值的TTL（较短，防止长期占用内存）
    NormalTTL     = 30 * time.Minute    // 正常数据的TTL
)

// NullCache 支持缓存空对象的缓存实现
type NullCache struct {
    rdb       *redis.Client
    nullTTL   time.Duration
    normalTTL time.Duration
}

func NewNullCache(rdb *redis.Client) *NullCache {
    return &NullCache{
        rdb:       rdb,
        nullTTL:   NullTTL,
        normalTTL: NormalTTL,
    }
}

// Get 获取数据（自动处理空值缓存）
func (c *NullCache) Get(
    ctx context.Context,
    key string,
    loadFn func(ctx context.Context) (interface{}, error),
) (interface{}, error) {
    // Step 1: 从缓存获取
    val, err := c.rdb.Get(ctx, key).Result()
  
    if err == nil {
        // 缓存命中
        // 检查是否是空值标记
        if val == NullValue {
            // 命中空值，直接返回"数据不存在"，不查数据库
            return nil, ErrDataNotFound
        }
        // 正常数据，反序列化返回
        var result interface{}
        json.Unmarshal([]byte(val), &result)
        return result, nil
    }

    if err != redis.Nil {
        return nil, err
    }

    // Step 2: 缓存未命中，查询数据库
    data, err := loadFn(ctx)
    if err != nil {
        return nil, err
    }

    // Step 3: 判断数据库查询结果
    if data == nil {
        // 数据库也不存在 → 缓存空值
        c.rdb.Set(ctx, key, NullValue, c.nullTTL)
        return nil, ErrDataNotFound
    }

    // 数据库存在 → 缓存正常数据
    dataBytes, _ := json.Marshal(data)
    c.rdb.Set(ctx, key, string(dataBytes), c.normalTTL)
    return data, nil
}

// 使用示例
func Example() {
    cache := NewNullCache(rdb)
  
    // 查询用户（用户不存在的情况）
    data, err := cache.Get(ctx, "user:999999", func(ctx context.Context) (interface{}, error) {
        return db.QueryUser(ctx, 999999) // 返回 nil, nil
    })
  
    if errors.Is(err, ErrDataNotFound) {
        // 数据不存在（已被缓存为空值，后续请求不再查DB）
        return "user not found"
    }
}

package cache

import (
    "context"
    "hash/fnv"
    "math"

    "github.com/redis/go-redis/v9"
)

// BloomFilter 基于 Redis 的布隆过滤器实现
type BloomFilter struct {
    rdb     *redis.Client
    key     string  // Redis 中存储位数组的 key
    bitSize uint    // 位数组大小
    hashNum uint    // 哈希函数个数
}

// NewBloomFilter 创建布隆过滤器
// expectedInsertions: 预计插入的元素数量
// falsePositiveRate: 期望的误判率（如 0.01 表示 1%）
func NewBloomFilter(
    rdb *redis.Client,
    key string,
    expectedInsertions uint64,
    falsePositiveRate float64,
) *BloomFilter {
    // 计算最优位数组大小: m = -(n * ln(p)) / (ln(2))^2
    bitSize := uint(math.Ceil(
        -float64(expectedInsertions) * math.Log(falsePositiveRate) /
            (math.Log(2) * math.Log(2)),
    ))

    // 计算最优哈希函数数量: k = (m/n) * ln(2)
    hashNum := uint(math.Ceil(
        (float64(bitSize) / float64(expectedInsertions)) * math.Log(2),
    ))

    return &BloomFilter{
        rdb:     rdb,
        key:     key,
        bitSize: bitSize,
        hashNum: hashNum,
    }
}

// Add 添加元素到布隆过滤器
func (bf *BloomFilter) Add(ctx context.Context, value string) error {
    pipe := bf.rdb.Pipeline()
  
    for i := uint(0); i < bf.hashNum; i++ {
        index := bf.hash(value, i)
        pipe.SetBit(ctx, bf.key, int64(index), 1)
    }
  
    _, err := pipe.Exec(ctx)
    return err
}

// AddBatch 批量添加元素
func (bf *BloomFilter) AddBatch(ctx context.Context, values []string) error {
    pipe := bf.rdb.Pipeline()
  
    for _, value := range values {
        for i := uint(0); i < bf.hashNum; i++ {
            index := bf.hash(value, i)
            pipe.SetBit(ctx, bf.key, int64(index), 1)
        }
    }
  
    _, err := pipe.Exec(ctx)
    return err
}

// Exists 检查元素是否可能存在
// 返回 true: 元素可能存在（需要继续查缓存/DB）
// 返回 false: 元素一定不存在（直接拒绝）
func (bf *BloomFilter) Exists(ctx context.Context, value string) (bool, error) {
    for i := uint(0); i < bf.hashNum; i++ {
        index := bf.hash(value, i)
        bit, err := bf.rdb.GetBit(ctx, bf.key, int64(index)).Result()
        if err != nil {
            return false, err
        }
        if bit == 0 {
            // 任一位为0，元素一定不存在
            return false, nil
        }
    }
    // 所有位都为1，元素可能存在
    return true, nil
}

// hash 使用 FNV 哈希 + 种子模拟多个哈希函数
func (bf *BloomFilter) hash(value string, seed uint) uint {
    h := fnv.New64a()
    h.Write([]byte(value))
    // 加入种子产生不同的哈希值
    h.Write([]byte{byte(seed)})
    return uint(h.Sum64()) % bf.bitSize
}

// InitFromDB 从数据库加载所有合法ID到布隆过滤器
func (bf *BloomFilter) InitFromDB(ctx context.Context, loadAllIDs func() ([]string, error)) error {
    ids, err := loadAllIDs()
    if err != nil {
        return err
    }
  
    // 批量添加到布隆过滤器
    batchSize := 1000
    for i := 0; i < len(ids); i += batchSize {
        end := i + batchSize
        if end > len(ids) {
            end = len(ids)
        }
        if err := bf.AddBatch(ctx, ids[i:end]); err != nil {
            return err
        }
    }
    return nil
}

package cache

import (
    "context"
    "errors"
    "fmt"
    "time"
)

var (
    ErrInvalidParam   = errors.New("invalid parameter")
    ErrDataNotExist   = errors.New("data does not exist")
    ErrServiceDegraded = errors.New("service degraded")
)

// PenetrationGuard 缓存穿透防护器（三层防护）
type PenetrationGuard struct {
    bloomFilter *BloomFilter // 第三层：布隆过滤器
    nullCache   *NullCache   // 第二层：空值缓存
}

// Get 带三层防护的缓存查询
func (g *PenetrationGuard) Get(
    ctx context.Context,
    key string,
    loadFn func(ctx context.Context) (interface{}, error),
) (interface{}, error) {
    // ========== 第一层：参数校验 ==========
    // （在调用此方法之前已完成，此处省略）
  
    // ========== 第二层：布隆过滤器检查 ==========
    exists, err := g.bloomFilter.Exists(ctx, key)
    if err != nil {
        // 布隆过滤器查询失败，降级处理（放行，让后续层处理）
        // 这里选择放行而不是拒绝，保证可用性
    }
    if !exists {
        // 布隆过滤器判定不存在 → 一定不存在 → 直接拒绝
        return nil, ErrDataNotExist
    }

    // ========== 第三层：空值缓存 ==========
    // 布隆过滤器说可能存在，再查缓存（含空值缓存）
    data, err := g.nullCache.Get(ctx, key, loadFn)
    if err != nil {
        if errors.Is(err, ErrDataNotFound) {
            // 数据库也不存在
            return nil, ErrDataNotExist
        }
        return nil, err
    }

    return data, nil
}

// ============ 完整使用示例 ============
func GetUserExample() {
    // 初始化
    guard := &PenetrationGuard{
        bloomFilter: NewBloomFilter(rdb, "bloom:users", 1000000, 0.01),
        nullCache:   NewNullCache(rdb),
    }

    // 启动时从数据库加载所有合法用户ID到布隆过滤器
    guard.bloomFilter.InitFromDB(ctx, func() ([]string, error) {
        return db.GetAllUserIDs()
    })

    // 处理请求
    userID := "-1" // 恶意请求
  
    // 第一层：参数校验
    if id, err := validateUserID(userID); err != nil {
        return "参数不合法" // 直接拒绝，-1 不合法
    }

    // 第二层+第三层：布隆过滤器 + 空值缓存
    data, err := guard.Get(ctx, "user:999999", func(ctx context.Context) (interface{}, error) {
        return db.GetUser(ctx, 999999)
    })
  
    if errors.Is(err, ErrDataNotExist) {
        return "用户不存在" // 布隆过滤器或空值缓存拦截
    }
}

// RebuildBloomFilter 定期重建布隆过滤器
func RebuildBloomFilter(ctx context.Context, bf *BloomFilter) error {
    // 使用新key创建新的布隆过滤器
    newKey := bf.key + ":new"
    newBF := NewBloomFilter(bf.rdb, newKey, bf.estimatedCapacity, bf.falsePositiveRate)
  
    // 从数据库加载最新数据
    err := newBF.InitFromDB(ctx, func() ([]string, error) {
        return db.GetAllValidIDs(ctx) // 只加载有效数据
    })
    if err != nil {
        return err
    }
  
    // 原子切换：rename 操作在 Redis 中是原子的
    // 将旧key备份，新key替换为正式key
    pipe := bf.rdb.Pipeline()
    pipe.Rename(ctx, bf.key, bf.key+":old")
    pipe.Rename(ctx, newKey, bf.key)
    _, err = pipe.Exec(ctx)
    if err != nil {
        return err
    }
  
    // 延迟删除旧key
    go func() {
        time.Sleep(1 * time.Minute)
        bf.rdb.Del(context.Background(), bf.key+":old")
    }()
  
    return nil
}

// StartPeriodicRebuild 启动定时重建任务
func StartPeriodicRebuild(ctx context.Context, bf *BloomFilter, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()
  
    for {
        select {
        case <-ticker.C:
            if err := RebuildBloomFilter(ctx, bf); err != nil {
                log.Printf("failed to rebuild bloom filter: %v", err)
            } else {
                log.Printf("bloom filter rebuilt successfully")
            }
        case <-ctx.Done():
            return
        }
    }
}