hotime/cache/cache_redis.go

package cache

import (
	. "code.hoteas.com/golang/hotime/common"
	"code.hoteas.com/golang/hotime/log"
	"github.com/garyburd/redigo/redis"
	"strings"
	"sync"
	"time"
)

type CacheRedis struct {
	TimeOut    int64
	DbSet      bool
	SessionSet bool
	Host       string
	Pwd        string
	Port       int64
	pool       *redis.Pool
	tag        int64
	ContextBase
	Log      *log.Logger
	initOnce sync.Once
}

func (that *CacheRedis) logErr(err error) {
	if err != nil && that.Log != nil {
		that.Log.Error().Err(err).Msg("redis cache error")
	}
}

// 唯一标志
func (that *CacheRedis) GetTag() int64 {

	if that.tag == int64(0) {
		that.tag = time.Now().UnixNano()
	}
	return that.tag
}

// initPool 初始化连接池（只执行一次）
func (that *CacheRedis) initPool() {
	that.initOnce.Do(func() {
		that.pool = &redis.Pool{
			MaxIdle:     10,              // 最大空闲连接数
			MaxActive:   100,             // 最大活跃连接数，0表示无限制
			IdleTimeout: 5 * time.Minute, // 空闲连接超时时间
			Wait:        true,            // 当连接池耗尽时是否等待
			Dial: func() (redis.Conn, error) {
				conn, err := redis.Dial("tcp", that.Host+":"+ObjToStr(that.Port),
					redis.DialConnectTimeout(5*time.Second),
					redis.DialReadTimeout(3*time.Second),
					redis.DialWriteTimeout(3*time.Second),
				)
				if err != nil {
					return nil, err
				}
				if that.Pwd != "" {
					if _, err := conn.Do("AUTH", that.Pwd); err != nil {
						conn.Close()
						return nil, err
					}
				}
				return conn, nil
			},
			TestOnBorrow: func(c redis.Conn, t time.Time) error {
				if time.Since(t) < time.Minute {
					return nil
				}
				_, err := c.Do("PING")
				return err
			},
		}
	})
}

// getConn 从连接池获取连接
func (that *CacheRedis) getConn() redis.Conn {
	that.initPool()
	if that.pool == nil {
		return nil
	}
	return that.pool.Get()
}

func (that *CacheRedis) del(key string) {
	conn := that.getConn()
	if conn == nil {
		return
	}
	defer conn.Close()

	del := strings.Index(key, "*")
	if del != -1 {
		val, err := redis.Strings(conn.Do("KEYS", key))
		if err != nil {
			that.logErr(err)
			return
		}
		if len(val) == 0 {
			return
		}
		conn.Send("MULTI")
		for i := range val {
			conn.Send("DEL", val[i])
		}
		_, err = conn.Do("EXEC")
		if err != nil {
			that.logErr(err)
		}
	} else {
		_, err := conn.Do("DEL", key)
		if err != nil {
			that.logErr(err)
		}
	}
}

// key value ,时间为时间戳
func (that *CacheRedis) set(key string, value string, expireSeconds int64) {
	conn := that.getConn()
	if conn == nil {
		return
	}
	defer conn.Close()

	_, err := conn.Do("SET", key, value, "EX", ObjToStr(expireSeconds))
	if err != nil {
		that.logErr(err)
	}
}

func (that *CacheRedis) get(key string) *Obj {
	reData := &Obj{}
	conn := that.getConn()
	if conn == nil {
		return reData
	}
	defer conn.Close()

	var err error
	reData.Data, err = redis.String(conn.Do("GET", key))
	if err != nil {
		reData.Data = nil
		if !strings.Contains(err.Error(), "nil returned") {
			that.logErr(err)
		}
	}
	return reData
}

func (that *CacheRedis) Cache(key string, data ...interface{}) *Obj {
	reData := &Obj{}

	//查询缓存
	if len(data) == 0 {
		reData = that.get(key)
		return reData
	}

	tim := int64(0)
	//删除缓存
	if len(data) == 1 && data[0] == nil {
		that.del(key)
		return reData
	}
	//添加缓存
	if len(data) == 1 {
		if that.TimeOut == 0 {
			//that.Time = Config.GetInt64("cacheShortTime")
		}
		tim += that.TimeOut
	}
	if len(data) == 2 {
		tempt := ObjToInt64(data[1])
		if tempt > tim {
			tim = tempt
		} else {
			tim = tim + tempt
		}
	}

	that.set(key, ObjToStr(data[0]), tim)

	return reData
}

// CachesGet 批量获取缓存（使用 Redis MGET 命令优化）
// 返回 Map，key 为缓存键，value 为缓存值（不存在的 key 不包含在结果中）
func (that *CacheRedis) CachesGet(keys []string) Map {
	result := make(Map, len(keys))
	if len(keys) == 0 {
		return result
	}

	conn := that.getConn()
	if conn == nil {
		return result
	}
	defer conn.Close()

	// 构建 MGET 参数
	args := make([]interface{}, len(keys))
	for i, key := range keys {
		args[i] = key
	}

	values, err := redis.Strings(conn.Do("MGET", args...))
	if err != nil {
		if !strings.Contains(err.Error(), "nil returned") {
			that.logErr(err)
		}
		return result
	}

	// 将结果映射回 Map
	for i, value := range values {
		if value != "" {
			result[keys[i]] = value
		}
	}

	return result
}

// CachesSet 批量设置缓存（使用 Redis pipeline 优化）
// data: Map，key 为缓存键，value 为缓存值
// timeout: 可选过期时间（秒），不传则使用默认超时时间
func (that *CacheRedis) CachesSet(data Map, timeout ...int64) {
	if len(data) == 0 {
		return
	}

	conn := that.getConn()
	if conn == nil {
		return
	}
	defer conn.Close()

	tim := that.TimeOut
	if len(timeout) > 0 && timeout[0] > 0 {
		if timeout[0] > tim {
			tim = timeout[0]
		} else {
			tim = tim + timeout[0]
		}
	}

	// 使用 pipeline 批量设置
	conn.Send("MULTI")
	for key, value := range data {
		conn.Send("SET", key, ObjToStr(value), "EX", ObjToStr(tim))
	}
	_, err := conn.Do("EXEC")
	if err != nil {
		that.logErr(err)
	}
}

// CachesDelete 批量删除缓存（使用 Redis DEL 命令批量删除）
func (that *CacheRedis) CachesDelete(keys []string) {
	if len(keys) == 0 {
		return
	}

	conn := that.getConn()
	if conn == nil {
		return
	}
	defer conn.Close()

	// 构建 DEL 参数
	args := make([]interface{}, len(keys))
	for i, key := range keys {
		args[i] = key
	}

	_, err := conn.Do("DEL", args...)
	if err != nil {
		that.logErr(err)
	}
}