eta_task/utils/redis/cluster_redis.go

package redis

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/go-redis/redis/v8"
	"strings"
	"time"
)

// ClusterRedisClient
// @Description: 集群的redis客户端
type ClusterRedisClient struct {
	redisClient *redis.ClusterClient
}

var DefaultKey = "zcmRedis"

// InitClusterRedis
// @Description: 初始化集群redis客户端
// @param config
// @return clusterRedisClient
// @return err
func InitClusterRedis(config string) (clusterRedisClient *ClusterRedisClient, err error) {
	var cf map[string]string
	err = json.Unmarshal([]byte(config), &cf)
	if err != nil {
		return
	}
	//if _, ok := cf["key"]; !ok {
	//	cf["key"] = DefaultKey
	//}

	// 集群地址
	connList := make([]string, 0)
	if _, ok := cf["conn"]; !ok {
		err = errors.New("config has no conn key")
		return
	} else {
		connList = strings.Split(cf["conn"], ",")
		if len(connList) <= 1 {
			err = errors.New("conn address less than or equal to 1")
			return
		}
	}

	// 密码
	if _, ok := cf["password"]; !ok {
		cf["password"] = ""
	}

	// 创建 Redis 客户端配置对象
	clusterOptions := &redis.ClusterOptions{
		Addrs:    connList, // 设置 Redis 节点的 IP 地址和端口号
		Password: cf["password"],
	}

	// 创建 Redis 集群客户端
	client := redis.NewClusterClient(clusterOptions)

	// 测试连接并获取信息
	_, err = client.Ping(context.TODO()).Result()
	if err != nil {
		err = errors.New("redis 链接失败:" + err.Error())
		return
	}

	clusterRedisClient = &ClusterRedisClient{redisClient: client}

	return
}

// Get
// @Description: 根据key获取数据（其实是返回的字节编码）
// @receiver rc
// @param key
// @return interface{}
func (rc *ClusterRedisClient) Get(key string) interface{} {
	data, err := rc.redisClient.Get(context.TODO(), key).Bytes()
	if err != nil {
		return nil
	}

	return data
}

// GetStr
// @Description: 根据key获取字符串数据
// @receiver rc
// @param key
// @return string
func (rc *ClusterRedisClient) GetStr(key string) string {
	return rc.redisClient.Get(context.TODO(), key).Val()
}

// GetInt
// @Description: 根据key获取int数据
// @receiver rc
// @param key
// @return int
// @return error
func (rc *ClusterRedisClient) GetInt64(key string) (int64, error) {
	return rc.redisClient.Get(context.TODO(), key).Int64()
}

// GetInt
// @Description: 根据key获取int数据
// @receiver rc
// @param key
// @return int
// @return error
func (rc *ClusterRedisClient) GetUInt64(key string) (uint64, error) {
	return rc.redisClient.Get(context.TODO(), key).Uint64()
}

// RedisBytes
// @Description: 根据key获取字节编码数据
// @receiver rc
// @param key
// @return data
// @return err
func (rc *ClusterRedisClient) RedisBytes(key string) (data []byte, err error) {
	data, err = rc.redisClient.Get(context.TODO(), key).Bytes()

	return
}

// RedisString
// @Description: 根据key获取字符串数据
// @receiver rc
// @param key
// @return data
// @return err
func (rc *ClusterRedisClient) RedisString(key string) (data string, err error) {
	data, err = rc.redisClient.Get(context.TODO(), key).Result()

	return
}

// RedisInt
// @Description: 根据key获取int数据
// @receiver rc
// @param key
// @return data
// @return err
func (rc *ClusterRedisClient) RedisInt(key string) (data int, err error) {
	data, err = rc.redisClient.Get(context.TODO(), key).Int()

	return
}

// Put
// @Description: put一个数据到redis
// @receiver rc
// @param key
// @param val
// @param timeout
// @return error
func (rc *ClusterRedisClient) Put(key string, val interface{}, timeout time.Duration) error {
	var err error
	err = rc.redisClient.SetEX(context.TODO(), key, val, timeout).Err()
	if err != nil {
		return err
	}

	err = rc.redisClient.HSet(context.TODO(), DefaultKey, key, true).Err()

	return err
}

// SetNX
// @Description: 设置一个会过期时间的值
// @receiver rc
// @param key
// @param val
// @param timeout
// @return bool
func (rc *ClusterRedisClient) SetNX(key string, val interface{}, timeout time.Duration) bool {
	result, err := rc.redisClient.SetNX(context.TODO(), key, val, timeout).Result()
	if err != nil {
		return false
	}

	return result
}

// Delete
// @Description: 删除redis中的键值对
// @receiver rc
// @param key
// @return error
func (rc *ClusterRedisClient) Delete(key string) error {
	var err error

	err = rc.redisClient.Del(context.TODO(), key).Err()
	if err != nil {
		return err
	}

	err = rc.redisClient.HDel(context.TODO(), DefaultKey, key).Err()

	return err
}

// IsExist
// @Description: 根据key判断是否写入缓存中
// @receiver rc
// @param key
// @return bool
func (rc *ClusterRedisClient) IsExist(key string) bool {
	result, err := rc.redisClient.Exists(context.TODO(), key).Result()
	if err != nil {
		return false
	}
	if result == 0 {
		_ = rc.redisClient.HDel(context.TODO(), DefaultKey, key).Err()
		return false
	}

	return true
}

// LPush
// @Description: 写入list
// @receiver rc
// @param key
// @param val
// @return error
func (rc *ClusterRedisClient) LPush(key string, val interface{}) error {
	data, _ := json.Marshal(val)
	err := rc.redisClient.LPush(context.TODO(), key, data).Err()

	return err
}

// Brpop
// @Description: 从list中读取
// @receiver rc
// @param key
// @param callback
func (rc *ClusterRedisClient) Brpop(key string, callback func([]byte)) {
	values, err := rc.redisClient.BRPop(context.TODO(), 1*time.Second, key).Result()
	if err != nil {
		return
	}
	if len(values) < 2 {
		fmt.Println("assert is wrong")
		return
	}

	callback([]byte(values[1]))

}

// GetRedisTTL
// @Description: 获取key的过期时间
// @receiver rc
// @param key
// @return time.Duration
func (rc *ClusterRedisClient) GetRedisTTL(key string) time.Duration {
	value, err := rc.redisClient.TTL(context.TODO(), key).Result()
	if err != nil {
		return 0
	}

	return value

}

// Incrby
// @Description: 设置自增值
// @receiver rc
// @param key
// @param num
// @return interface{}
// @return error
func (rc *ClusterRedisClient) Incrby(key string, num int) (interface{}, error) {
	return rc.redisClient.IncrBy(context.TODO(), key, int64(num)).Result()
}

// Do
// @Description: cmd执行redis命令
// @receiver rc
// @param commandName
// @param args
// @return reply
// @return err
func (rc *ClusterRedisClient) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
	newArgs := []interface{}{commandName}
	newArgs = append(newArgs, args...)
	return rc.redisClient.Do(context.TODO(), newArgs...).Result()
}