分布式多级缓存

https://github.com/gin-contrib/cache/master/blob/persistence/cache_store.go
type CacheStore interface {
    // Get retrieves an item from the cache. Returns the item or nil, and a bool indicating
    // whether the key was found.
    Get(key string, value interface{}) error

    // Set sets an item to the cache, replacing any existing item.
    Set(key string, value interface{}, expire time.Duration) error

    // Add adds an item to the cache only if an item doesn't already exist for the given
    // key, or if the existing item has expired. Returns an error otherwise.
    Add(key string, value interface{}, expire time.Duration) error

    // Replace sets a new value for the cache key only if it already exists. Returns an
    // error if it does not.
    Replace(key string, data interface{}, expire time.Duration) error

    // Delete removes an item from the cache. Does nothing if the key is not in the cache.
    Delete(key string) error

    // Increment increments a real number, and returns error if the value is not real
    Increment(key string, data uint64) (uint64, error)

    // Decrement decrements a real number, and returns error if the value is not real
    Decrement(key string, data uint64) (uint64, error)

    // Flush deletes all items from the cache.
    Flush() error
}

// https://github.com/mgtv-tech/jetcache-go/blob/7570bd5d950b4e016cf241b45e3536c50e49cd56/cache.go
type (
    // Cache interface is used to define the cache implementation.
    Cache interface {
        // Set sets the cache with ItemOption
        Set(ctx context.Context, key string, opts ...ItemOption) error
        // Once gets the opts.value for the given key from the cache or
        // executes, caches, and returns the results of the given opts.do,
        // making sure that only one execution is in-flight for a given key
        // at a time. If a duplicate comes in, the duplicate caller waits for the
        // original to complete and receives the same results.
        Once(ctx context.Context, key string, opts ...ItemOption) error
        // Delete deletes cached val with key.
        Delete(ctx context.Context, key string) error
        // DeleteFromLocalCache deletes local cached val with key.
        DeleteFromLocalCache(key string)
        // Exists reports whether val for the given key exists.
        Exists(ctx context.Context, key string) bool
        // Get gets the val for the given key and fills into val.
        Get(ctx context.Context, key string, val any) error
        // GetSkippingLocal gets the val for the given key skipping local cache.
        GetSkippingLocal(ctx context.Context, key string, val any) error
        // TaskSize returns Refresh task size.
        TaskSize() int
        // CacheType returns cache type
        CacheType() string
        // Close closes the cache. This should be called when cache refreshing is
        // enabled and no longer needed, or when it may lead to resource leaks.
        Close()
    }

    jetCache struct {
        sync.Mutex // 分布式锁
        Options  // 选择函数
        group          singleflight.Group // 避免同时执行重复的函数
        safeRand       *util.SafeRand     // 线程安全的随机数生成器
        refreshTaskMap sync.Map           // 更新任务map
        eventCh        chan *Event        // 事件管道
        stopChan       chan struct{}      // 停止管道
    }
)

type (
    // Options are used to store cache options.
    Options struct {
        name                       string             // Cache name, used for log identification and metric reporting
        remote                     remote.Remote      // Remote is distributed cache, such as Redis.
        local                      local.Local        // Local is memory cache, such as FreeCache.
        codec                      string             // Value encoding and decoding method. Default is "msgpack.Name". You can also customize it.
        errNotFound                error              // Error to return for cache miss. Used to prevent cache penetration.
        remoteExpiry               time.Duration      // Remote cache ttl, Default is 1 hour.
        notFoundExpiry             time.Duration      // Duration for placeholder cache when there is a cache miss. Default is 1 minute.
        offset                     time.Duration      // Expiration time jitter factor for cache misses.
        refreshDuration            time.Duration      // Interval for asynchronous cache refresh. Default is 0 (refresh is disabled).
        stopRefreshAfterLastAccess time.Duration      // Duration for cache to stop refreshing after no access. Default is refreshDuration + 1 second.
        refreshConcurrency         int                // Maximum number of concurrent cache refreshes. Default is 4.
        statsDisabled              bool               // Flag to disable cache statistics.
        statsHandler               stats.Handler      // Metrics statsHandler collector.
        sourceID                   string             // Unique identifier for cache instance.
        syncLocal                  bool               // Enable events for syncing local cache (only for "Both" cache type).
        eventChBufSize             int                // Buffer size for event channel (default: 100).
        eventHandler               func(event *Event) // Function to handle local cache invalidation events.
    }

    // Option defines the method to customize an Options.
    Option func(o *Options)  // Option 修改函数

)


func WithName(name string) Option {
    return func(o *Options) {
        o.name = name
    }
}

func WithRemote(remote remote.Remote) Option {
    return func(o *Options) {
        o.remote = remote
    }
}

func WithLocal(local local.Local) Option {
    return func(o *Options) {
        o.local = local
    }
}
func newOptions(opts ...Option) Options {
    var o Options
    for _, opt := range opts {
        opt(&o)
    }
}

// 使用代码示例如下：
// o := newOptions(
//    WithOffset(30 * time.Second),
//    WithRefreshConcurrency(16),
//    WithCodec(json.Name),
// )

func Example_advancedUsage() {
    // 创建redis ring
    ring := redis.NewRing(&redis.RingOptions{
        Addrs: map[string]string{
        "localhost": ":6379",
        },
    })
    // 创建cache
    mycache := cache.New(
        cache.WithName("any"),                                          // 指定名称
        cache.WithRemote(remote.NewGoRedisV8Adaptor(ring)),             // 添加远程redis缓存
        cache.WithLocal(local.NewFreeCache(256*local.MB, time.Minute)), // 添加本地缓存
        cache.WithErrNotFound(errRecordNotFound),                       // 指定无数据异常
        cache.WithRefreshDuration(time.Minute),                         // 指定刷新时间
    )

    ctx := context.TODO()
    key := "mykey:1"
    obj := new(object)
    // 查询对应值
    if err := mycache.Once(
        ctx,
        key,
        cache.Value(obj),     // 设置值类型
        cache.TTL(time.Hour), // 设置TTL时间
        cache.Refresh(true),  // 是否进行刷新
        cache.Do(
        func(ctx context.Context) (any, error) {
            return mockDBGetObject(1)
        }), // 值处理函数--当出现缓存not found 时调用此函数
    ); err != nil {
        panic(err)
    }
    // 输出最终对象
    fmt.Println(obj)
    // Output: &{mystring 42}
    // 关闭缓存块
    mycache.Close()
}

func (c *jetCache) getSetItemBytesOnce(item *item) (b []byte, cached bool, err error) {
    if !item.skipLocal && c.local != nil {
        // 先统local 进行值查询
        b, ok := c.local.Get(item.key)
        // 更新对应命中率
        if ok {
            c.statsHandler.IncrHit()
            c.statsHandler.IncrLocalHit()
            if bytes.Compare(b, notFoundPlaceholder) == 0 {
                return nil, true, c.errNotFound
            }
            return b, true, nil
        }
    }
    // 使用group.Do 保证一次只有一个再执行
    v, err, _ := c.group.Do(
        item.key, // key 用于锁住函数
        // 核心的执行函数
        func() (any, error) {
            // 查询对应值--
            b, err := c.getBytes(item.Context(), item.key, item.skipLocal)
            if err == nil {
                cached = true
                return b, nil
            } else if errors.Is(err, c.errNotFound) {
                cached = true
                return nil, c.errNotFound
            }
            // 缓存中不存在该值--缓存穿透了--执行对应是饥饿中加密手机
            b, ok, err := c.set(item)
            if ok {
                c.send(EventTypeSetByOnce, item.key)
                return b, nil
            }

        return nil, err
    })

    if err != nil {
        return nil, false, err
    }

    return v.([]byte), cached, nil
}

func (c *jetCache) set(item *item) ([]byte, bool, error) {
    // 先查询对象是否存在
    val, err := item.getValue()
    // 设置函数非空，新增参数
    if item.do != nil {
        c.statsHandler.IncrQuery()
    }

    if c.IsNotFound(err) {
        if e := c.setNotFound(item.Context(), item.key, item.skipLocal); e != nil {
        logger.Error("setNotFound(%s) error(%v)", item.key, err)
        }
        return notFoundPlaceholder, true, nil
    } else if err != nil {
        c.statsHandler.IncrQueryFail(err)
        return nil, false, err
    }

    b, err := c.Marshal(val)
    if err != nil {
        return nil, false, err
    }
    // 设置本地缓存
    if c.local != nil && !item.skipLocal {
        c.local.Set(item.key, b)
    }

    if c.remote == nil {
        if c.local == nil {
            return b, true, ErrRemoteLocalBothNil
        }
        return b, true, nil
    }
    // 查询剩余存在时间
    ttl := item.getTtl(c.remoteExpiry)
    if ttl == 0 {
        return b, true, nil
    }
    // 设置远端缓存
    if item.setXX {
        _, err := c.remote.SetXX(item.Context(), item.key, b, ttl)
        return b, true, err
    }
    if item.setNX {
        _, err := c.remote.SetNX(item.Context(), item.key, b, ttl)
        return b, true, err
    }
    return b, true, c.remote.SetEX(item.Context(), item.key, b, ttl)
}

func (item *item) getValue() (any, error) {
    if item.do != nil {
        // 执行数据处理--使用用户注入钩子进行处理
        return item.do(item.Context())
    }
    if item.value != nil {
        return item.value, nil
    }
    return nil, nil
}