package cache import ( "context" "errors" "fmt" "time" "fwh-toolbox-server/cache/redisx" "github.com/redis/go-redis/v9" ) func Init(addr string) { opt := redisx.Config{ Addrs: []string{addr}, DB: 12, } redisx.Init(&opt) } // ErrNotFound 用于表示缓存未命中 var ErrNotFound = errors.New("cachex: not found") // helper: compute full key with prefix func fullKey(prefix, key string) string { if prefix == "" { return key } return fmt.Sprintf("%s:%s", prefix, key) } // RegisterHitHook / RegisterMissHook: 简单 hook,用户可利用来统计 var ( hitHook func(key string) missHook func(key string) ) func RegisterHitHook(fn func(key string)) { hitHook = fn } func RegisterMissHook(fn func(key string)) { missHook = fn } // internal helpers to call hooks safely func onHit(key string) { if hitHook != nil { go func() { defer func() { recover() }(); hitHook(key) }() } } func onMiss(key string) { if missHook != nil { go func() { defer func() { recover() }(); missHook(key) }() } } /********* 字符串操作 *********/ // Set string value func Set(ctx context.Context, key string, value interface{}, ttl time.Duration) error { return redisx.Set(ctx, key, value, ttl) } // Get string value func Get(ctx context.Context, key string) (string, error) { val, err := redisx.Get(ctx, key) if err != nil { return "", err } return val, nil } func MSet(ctx context.Context, pairs ...interface{}) error { return redisx.MSet(ctx, pairs...) } func MGet(ctx context.Context, keys ...string) ([]interface{}, error) { return redisx.MGet(ctx, keys...) } func Incr(ctx context.Context, key string) (int64, error) { return redisx.Incr(ctx, key) } /********* 哈希操作 *********/ func HSet(ctx context.Context, key string, values ...interface{}) error { return redisx.HSet(ctx, key, values...) } func HGet(ctx context.Context, key, field string) (string, error) { return redisx.HGet(ctx, key, field) } func HGetAll(ctx context.Context, key string) (map[string]string, error) { return redisx.HGetAll(ctx, key) } /********* 列表操作 *********/ func LPush(ctx context.Context, key string, values ...interface{}) error { return redisx.LPush(ctx, key, values...) } func RPush(ctx context.Context, key string, values ...interface{}) error { return redisx.RPush(ctx, key, values...) } func LPop(ctx context.Context, key string) (string, error) { return redisx.LPop(ctx, key) } /********* 集合操作 *********/ func SAdd(ctx context.Context, key string, members ...interface{}) error { return redisx.SAdd(ctx, key, members...) } func SRem(ctx context.Context, key string, members ...interface{}) error { return redisx.SRem(ctx, key, members...) } func SMembers(ctx context.Context, key string) ([]string, error) { return redisx.SMembers(ctx, key) } func SIsMember(ctx context.Context, key string, member interface{}) (bool, error) { return redisx.SIsMember(ctx, key, member) } /********* 有序集合操作 *********/ func ZAdd(ctx context.Context, key string, members ...redis.Z) error { return redisx.ZAdd(ctx, key, members...) } func ZRange(ctx context.Context, key string, start, stop int64) ([]string, error) { return redisx.ZRange(ctx, key, start, stop) } /********* Stream(流)操作 *********/ // 添加消息 func XAdd(ctx context.Context, stream string, values map[string]interface{}) (string, error) { return redisx.XAdd(ctx, stream, values) } // 读取消息 func XRead(ctx context.Context, streams map[string]string, count int64) ([]redis.XStream, error) { return redisx.XRead(ctx, streams, count) } // 创建消费者组 func XGroupCreate(ctx context.Context, stream, group, start string) error { return redisx.XGroupCreate(ctx, stream, group, start) } // 从消费者组读取 func XReadGroup(ctx context.Context, group, consumer string, streams map[string]string, count int64) ([]redis.XStream, error) { return redisx.XReadGroup(ctx, group, consumer, streams, count) } // 从消费者组读取滞留消息 func XAutoClaim(ctx context.Context, stream string, group string, consumer string, idle time.Duration, count int64) (messages []redis.XMessage, start string, err error) { return redisx.XAutoClaim(ctx, stream, group, consumer, idle, count) } // 从消费者组读取指定ID的Pending状态信息 func XPendingExtOne(ctx context.Context, stream string, group string, consumer string, id string) (*redis.XPendingExt, error) { return redisx.XPendingExtOne(ctx, stream, group, consumer, id) } // 确认消息 func XAck(ctx context.Context, stream, group string, ids ...string) error { return redisx.XAck(ctx, stream, group, ids...) } // 确认同时删除消息 func XAckDel(ctx context.Context, stream, group string, ids ...string) error { return redisx.XAckDel(ctx, stream, group, ids...) } // 查询范围内指定的消息信息 func XRangeN(ctx context.Context, stream, start, stop string, count int64) ([]redis.XMessage, error) { return redisx.XRangeN(ctx, stream, start, stop, count) } // 获取完整的Stream信息 func XInfoStreamFull(ctx context.Context, key string, count int) (*redis.XInfoStreamFull, error) { return redisx.XInfoStreamFull(ctx, key, count) } /********* 分布式锁操作 *********/ type RedisLock = redisx.RedisLock func TryLock(ctx context.Context, key string, ttl time.Duration) (*RedisLock, bool, error) { return redisx.TryLock(ctx, key, ttl) } func Lock(ctx context.Context, key string, ttl, retryInterval, timeout time.Duration) (*RedisLock, error) { return redisx.Lock(ctx, key, ttl, retryInterval, timeout) } /********* JSON操作 *********/ // SetJSON 将任意结构体序列化为 JSON 存入 Redis func SetJSON(ctx context.Context, key string, value interface{}, ttl time.Duration) error { return redisx.SetJSON(ctx, key, value, ttl) } // GetJSON 从 Redis 获取并反序列化为指定结构体 func GetJSON[T any](ctx context.Context, key string, dest *T) error { return redisx.GetJSON[T](ctx, key, dest) } // MustGetJSON 返回默认值版本(未命中时返回 nil) func MustGetJSON[T any](ctx context.Context, key string) *T { return redisx.MustGetJSON[T](ctx, key) } // HSetJSON 在 Hash 中存储 JSON func HSetJSON(ctx context.Context, key, field string, value interface{}) error { return redisx.HSetJSON(ctx, key, field, value) } // HGetJSON 从 Hash 中读取 JSON func HGetJSON[T any](ctx context.Context, key, field string, dest *T) error { return redisx.HGetJSON[T](ctx, key, field, dest) } // MGetJSON 支持批量获取多个 JSON 对象 func MGetJSON[T any](ctx context.Context, keys ...string) (map[string]*T, []string, error) { return redisx.MGetJSON[T](ctx, keys...) } /********* Lua脚本操作 *********/ func RunScript(ctx context.Context, script string, keys []string, args ...interface{}) (interface{}, error) { return redisx.RunScript(ctx, script, keys, args...) } /********* 发布订阅操作 *********/ func Publish(ctx context.Context, channel string, msg interface{}) error { return redisx.Publish(ctx, channel, msg) } func Subscribe(ctx context.Context, channel string) *redis.PubSub { return redisx.Subscribe(ctx, channel) } /********* 管道与事务 *********/ func Pipeline(ctx context.Context, fn func(redis.Pipeliner) error) ([]redis.Cmder, error) { return redisx.Pipeline(ctx, fn) } /********* 工具方法 *********/ // Delete func Delete(ctx context.Context, keys ...string) error { return redisx.Del(ctx, keys...) } // Expire func Expire(ctx context.Context, key string, ttl time.Duration) (bool, error) { return redisx.Expire(ctx, key, ttl) } // ExpireAt func ExpireAt(ctx context.Context, key string, tm time.Time) (bool, error) { return redisx.ExpireAt(ctx, key, tm) } // TTL func TTL(ctx context.Context, key string) (time.Duration, error) { return redisx.TTL(ctx, key) } // Exists func Exists(ctx context.Context, key string) (bool, error) { return redisx.Exists(ctx, key) }