redis_v2/redis/goredis.go

package redis

import (
	"context"
	"encoding/json"
	"errors"
	"hash/crc32"
	"log"
	"math"
	"regexp"
	"strconv"
	"strings"
	"time"

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

//goredis 工具类对象
type GoRedis struct {
	//Password string
	Code   string
	DB     int   //默认库
	DBS    []int //数据库列表
	Nodes  []int //节点列表
	HashDb int   //是否是多个数据库 0单库 1单节点多库 2多节点单库 3多节点多库
	Ctx    context.Context
	CMap   map[int]map[int]*redis.Client
	BakDb  bool     //是否有备用节点
	Bak    *GoRedis //备用节点连接
}

//初始化单个"[other=]127.0.0.1:2203[|#]127.0.0.1:2204[/0]=0-1=1-10=300" [代码]、地址[|备用地址[/代表默认库][#多节点地址，默认库都为0]、库、最大池、空闲时间默认300秒
/*
示例
127.0.0.1:2203
127.0.0.1:2203=2=15  默认库为2，最大连接为15，最小连接是0.2*15为3
other=127.0.0.1:2203 带code配置
other=127.0.0.1:2203=0=10
other=127.0.0.1:2203=0=10=300 空闲时间300秒
other=127.0.0.1:2203=0-8=10=300 带code配置，使用多个库即0 1 2 3 4 5 6 7 8，最大连接为10，最小连接是0.2*10为2,空闲时间300秒
other=127.0.0.1:2203=0-8=1-10=300 带code配置，使用多个库即0 1 2 3 4 5 6 7 8，最大连接为10，最小连接1,空闲时间300秒
other=127.0.0.1:2203|127.0.0.1:2204=0-8=10=300 带code配置，使用备用节点（仅供取数据时使用），使用多个库即0 1 2 3 4 5 6 7 8，最大连接为10，最小连接1,空闲时间300秒
other=127.0.0.1:2203|127.0.0.1:2204/0=0-8=10=300 带code配置，使用备用节点（仅供取数据时使用，备用节点默认库为0），使用多个库即0 1 2 3 4 5 6 7 8，最大连接为10，最小连接1,空闲时间300秒
other=127.0.0.1:2203#127.0.0.1:2204#127.0.0.1:2205#127.0.0.1:2206=0=10=300 带code配置，使用多节点模式（多节点所有默认库为0），最大连接为10，最小连接2,空闲时间300秒
注意：
1、当使用多节点时，用正则获取keys时，再用Get取模获取数据时，一定要保障放入和取出是同一个hashcode，否则数据永远取不到
*/

//解析配置并初始化，opt为string或map
func (r *GoRedis) Init(opt interface{}) {
	check := false
	//代码  地址 []库范围 []池配置 空闲时间
	code, addr, dbs, pool, idle := "", "", []int{0}, []int{2, 30}, 300
	if so, ok := opt.(string); ok {
		arr := strings.Split(so, "=")
		regAddr := regexp.MustCompile("[0-9.a-zA-Z/]+:[0-9]+.*")
		if len(arr) == 1 { //只是一个串
			if regAddr.MatchString(arr[0]) {
				check = true
				addr = arr[0]
			}
		} else if len(arr) > 1 {
			index := 0
			if regAddr.MatchString(arr[0]) { //第1个是地址
				index = 1
				addr = arr[0]
				check = true
			} else if regAddr.MatchString(arr[1]) { //第二个是地址
				check = true
				addr = arr[1]
				code = arr[0]
			}
			//解析库配置
			if len(arr) > 2-index { //dbs配置
				dbs1 := strings.Split(arr[2-index], "-") //库范围的配置
				if len(dbs1) == 1 || len(dbs1) == 2 {
					check = true
					dbs[0], _ = strconv.Atoi(dbs1[0])
					if len(dbs1) == 2 {
						tmp, _ := strconv.Atoi(dbs1[1])
						dbs = append(dbs, tmp)
					}
				} else {
					check = false
				}
				//解析连接池配置
				if len(arr) > 3-index {
					pool1 := strings.Split(arr[3-index], "-")
					if len(pool1) == 1 || len(pool1) == 2 {
						check = true
						if len(pool1) == 1 {
							pool[1], _ = strconv.Atoi(pool1[0])
							pool[0] = int(math.Ceil(float64(pool[1]) * 0.2))
						} else {
							pool[0], _ = strconv.Atoi(pool1[0])
							pool[1], _ = strconv.Atoi(pool1[1])
						}
					} else {
						check = false
					}
					//解析最大空闲时间配置
					if len(arr) > 4-index {
						idle, _ = strconv.Atoi(arr[4-index])
						if idle == 0 {
							idle = 300
						}
					}
				}
			}
		}
	} else if _, ok := opt.(map[string]interface{}); ok {

	}
	if check {
		log.Println("代码  地址 []库范围 []池配置 空闲时间", code, addr, dbs, pool, idle)
		addrs := strings.Split(addr, "|") //备用节点模式
		if len(addrs) == 2 {              //备用节点的模式 2选1
			r.init(addrs[0], code, dbs, pool[1], pool[0], idle)
			r.BakDb = true
			//有备用节点，集群先不考虑，支持指定的库
			addr1 := strings.Split(addrs[1], "/")
			r.Bak = &GoRedis{}
			if len(addr1) == 2 { //没有指定库 根据/区分
				i, _ := strconv.Atoi(addr1[1])
				dbs = []int{i}
			}
			r.Bak.init(addr1[0], code, dbs, pool[1], pool[0], idle)
		} else if len(addrs) == 1 {
			addr1 := strings.Split(addrs[0], "#") //是多节点模式,所有库默认为0
			if len(addr1) == 1 {
				r.init(addr1[0], code, dbs, pool[1], pool[0], idle)
			} else { //多节点模式
				r.Code = code
				r.HashDb = 1 + len(dbs)
				if len(dbs) > 1 {
					r.DBS = []int{}
					for i := dbs[0]; i <= dbs[1]; i++ {
						r.DBS = append(r.DBS, i)
					}
				} else {
					r.DBS = dbs
				}
				r.DB = dbs[0]
				r.Ctx = context.Background()
				r.CMap = map[int]map[int]*redis.Client{}
				r.Nodes = []int{}
				for i := 0; i < len(addr1); i++ {
					r.Nodes = append(r.Nodes, i)
					r.CMap[i] = map[int]*redis.Client{}
					for k, v := range r.DBS { //按节点的每个库初始化
						r.CMap[i][k] = redis.NewClient(&redis.Options{
							Addr:         addr1[i],
							DB:           v,
							PoolSize:     pool[1],
							MinIdleConns: pool[0],
							IdleTimeout:  time.Duration(idle) * time.Second,
						})
					}
				}
			}
		}
	}
	log.Println(check, code, addr, dbs, pool, idle, r.DBS)
}

//初始化连接
func (r *GoRedis) init(addr, code string, dbs []int, poolSize, minIdleConns, idleTimeOut int) {
	r.Code = code
	r.DB = dbs[0]
	r.Ctx = context.Background()
	r.CMap = map[int]map[int]*redis.Client{}
	r.CMap[0] = map[int]*redis.Client{}
	if len(dbs) > 1 {
		r.HashDb = len(dbs) - 1
		r.DBS = []int{}
		for i := dbs[0]; i <= dbs[1]; i++ {
			r.DBS = append(r.DBS, i)
		}
	} else {
		r.DBS = dbs
	}
	for k, v := range r.DBS { //按节点的每个库初始化
		r.CMap[0][k] = redis.NewClient(&redis.Options{
			Addr:         addr,
			DB:           v,
			PoolSize:     poolSize,
			MinIdleConns: minIdleConns,
			IdleTimeout:  time.Duration(idleTimeOut) * time.Second,
		})
	}
}

//int转timeDuration
func D(t int) time.Duration {
	return time.Duration(t) * time.Second
}

//取db
func (r *GoRedis) GetDB(key string) (int, int) {
	switch r.HashDb {
	case 0: //单节点单库
		return 0, 0
	case 1: //单节点多库
		return 0, hashCode(key) % len(r.DBS)
	case 2: //多节点单库
		return hashCode(key[len(key)/2:]) % len(r.Nodes), 0
	case 3: ////多节点多库
		return hashCode(key[len(key)/2:]) % len(r.Nodes), hashCode(key) % len(r.DBS)
	}
	// if r.HashDb {
	// 	//return int(key[len(key)-1]) % len(r.DBS)
	// 	return hashCode(key) % len(r.DBS)
	// } else {
	// 	return r.DB
	// }
	return 0, 0
}

//根据key取hash
func hashCode(key string) int {
	v := int(crc32.ChecksumIEEE([]byte(key)))
	if v < 0 {
		v = -v
	}
	return v
}

func (r *GoRedis) Client(key string) *redis.Client {
	_i, _k := r.GetDB(key)
	m := r.CMap[_i]
	if m != nil {
		c := m[_k]
		if c != nil {
			return c
		}
	}
	return &redis.Client{}
}

//-----具体方法
//简单的Put方法
func (r *GoRedis) Put(key string, val interface{}) (string, error) {
	stutsCmd := r.Client(key).Set(r.Ctx, key, val, 0)
	str, err := stutsCmd.Result()
	return str, err
}

//存key，加过期时间
func (r *GoRedis) Set(key string, val interface{}, timeout int) (string, error) {
	stutsCmd := r.Client(key).Set(r.Ctx, key, val, D(timeout))
	//cmd := r.CMap[r.GetDB(key)].Do(r.Ctx, "setex", key, timeout, val)
	str, err := stutsCmd.Result()
	return str, err
}

//简单的Get方法,无结果返回空串，err: redis: nil
func (r *GoRedis) Get(key string) (string, error) {
	stutsCmd := r.Client(key).Get(r.Ctx, key)
	str, err := stutsCmd.Result()
	// log.Println("1", str, "-", err)
	if err != nil {
		//有备用节点
		if r.BakDb {
			str, err = r.Bak.Get(key)
		}
	}
	return str, err
}

//根据正则取key，未考虑负载、多库
func (r *GoRedis) GetByPattern(key string) (res []string, err error) {
	serr := ""
	for _, v1 := range r.CMap {
		for _, v := range v1 {
			strSlice := v.Keys(r.Ctx, key)
			arr, err1 := strSlice.Result()
			if len(arr) > 0 {
				res = append(res, arr...)
			} else if err1 != nil {
				serr += err1.Error()
			}
		}
	}
	if serr != "" {
		err = errors.New(serr)
	}
	return
}

//批量保存数据
//   不支持- MSet("key1", "value1", "key2", "value2")
//   不支持- MSet([]string{"key1", "value1", "key2", "value2"})
//  []interface{[]interface{k1,v1},[]interface{k2,v2},[]interface{k3,v3} }
//   - MSet(map[string]interface{}{"key1": "value1", "key2": "value2"})
func (r *GoRedis) BulkPut(timeout int, obj interface{}) {
	if r.HashDb > 0 {
		if timeout < 1 {
			timeout = 0
		}
		timeEx := time.Duration(timeout) * time.Second
		switch obj.(type) {
		case []interface{}:
			if objs, ok := obj.([]interface{}); ok {
				for _, _tmp := range objs {
					tmp, ok := _tmp.([]interface{})
					if ok && len(tmp) == 2 {
						key, ok1 := tmp[0].(string)
						if ok1 {
							r.Client(key).Set(r.Ctx, key, tmp[1], timeEx)
						}
					}
				}
			}
		case map[string]interface{}:
			if objs, ok := obj.(map[string]interface{}); ok {
				for k, v := range objs {
					r.Client(k).Set(r.Ctx, k, v, timeEx)
				}
			}
		}
	} else {
		//单库直接保存
		if timeout < 1 {
			stutsCmd := r.CMap[0][0].MSet(r.Ctx, obj)
			str, err := stutsCmd.Result()
			log.Println(str, err)
		} else {
			timeEx := time.Duration(timeout) * time.Second
			//设置超时
			lenth := 0
			cmds, err := r.CMap[0][0].Pipelined(r.Ctx, func(pipe redis.Pipeliner) error {
				if objs, ok := obj.([]interface{}); ok {
					lenth = len(objs)
					for _, _tmp := range objs {
						tmp, ok := _tmp.([]interface{})
						if ok && len(tmp) == 2 {
							key, ok1 := tmp[0].(string)
							if ok1 {
								pipe.Set(r.Ctx, key, tmp[1], timeEx)
							}
						}
					}

				} else if objs, ok := obj.(map[string]interface{}); ok {
					lenth = len(objs)
					for k, v := range objs {
						pipe.Set(r.Ctx, k, v, timeEx)
					}
				} else {
					log.Println("bulkPut type error")
				}
				return nil
			})
			if err != nil {
				log.Println("bulkPut error", err.Error())
			}
			if len(cmds) == lenth { //数据相等
				for _, v := range cmds {
					log.Println("cmd", v.Args(), v.String())
				}
			}
		}
	}
}

//设置超时时间,单位秒
func (r *GoRedis) SetExpire(key string, expire int) error {
	boolCmd := r.Client(key).Expire(r.Ctx, key, D(expire))
	return boolCmd.Err()
}

//判断一个key是否存在
func (r *GoRedis) Exists(key string) bool {
	intCmd := r.Client(key).Exists(r.Ctx, key)
	i, err := intCmd.Result()
	if err != nil {
		log.Println("redisutil-exists", key, err)
	}
	return i == 1
}

//直接返回字节流
func (r *GoRedis) GetBytes(key string) (ret *[]byte, err error) {
	cmd := r.Client(key).Do(r.Ctx, "GET", key)
	res, err := cmd.Result()
	log.Println(res)
	if err != nil {
		log.Println("redisutil-GetBytesError", err)
	} else {
		if tmp, ok := res.([]byte); ok {
			ret = &tmp
		} else if tmp, ok := res.(string); ok {
			bs := []byte(tmp)
			ret = &bs
		} else {
			err = errors.New("redis返回数据格式不对")
		}
	}
	return
}

//支持删除多个key
func (r *GoRedis) Del(key ...string) (b bool) {
	i := 0
	if r.HashDb > 0 {
		for _, k := range key {
			cmd := r.Client(k).Del(r.Ctx, k)
			i1, _ := cmd.Result()
			i += int(i1)
		}
	} else {
		intCmd := r.CMap[0][0].Del(r.Ctx, key...)
		i1, _ := intCmd.Result()
		i = int(i1)
	}
	return i == len(key)
}

//根据代码和前辍key删除多个
func (r *GoRedis) DelByPattern(key string) {
	res, _ := r.GetByPattern(key)
	if len(res) > 0 {
		r.Del(res...)
	}
}

//自增计数器
func (r *GoRedis) Incr(key string) (int64, error) {
	intCmd := r.Client(key).Incr(r.Ctx, key)
	i, err := intCmd.Result()
	return i, err
}

//自减
func (r *GoRedis) Decrby(key string, val int) (int64, error) {
	intCmd := r.Client(key).DecrBy(r.Ctx, key, int64(val))
	i, err := intCmd.Result()
	return i, err
}

//批量取多个key
func (r *GoRedis) Mget(key []string) []interface{} {
	if r.HashDb > 0 {
		mdb := map[int]map[int][]string{}
		for _, v := range key { //分组
			_i, _k := r.GetDB(v)
			mdb[_i] = map[int][]string{}
			arr := mdb[_i][_k]
			if arr == nil {
				arr = []string{v}
			} else {
				arr = append(arr, v)
			}
			mdb[_i][_k] = arr
		}
		res := []interface{}{}
		for k, v := range mdb {
			for k1, v1 := range v {
				sliceCmd := r.CMap[k][k1].MGet(r.Ctx, v1...)
				res1, err := sliceCmd.Result()
				if err != nil {
					log.Println("Mget error", err)
				}
				res = append(res, res1...)
			}
		}
		return res
	} else {
		sliceCmd := r.CMap[0][0].MGet(r.Ctx, key...)
		res, err := sliceCmd.Result()
		if err != nil {
			log.Println("Mget error", err)
		}
		return res
	}
}

//取出并删除Key
func (r *GoRedis) Pop(key string) (result interface{}) {
	c := r.Client(key)
	strCmd := c.Get(r.Ctx, key)
	b, err := strCmd.Bytes()
	if err != nil {
		log.Println("Poperr bytes", err)
		return
	}
	err1 := json.Unmarshal(b, &result)
	if err1 != nil {
		log.Println("Poperr json ", err)
		return
	} else {
		go c.Del(r.Ctx, key)
	}

	return
}

//list操作
func (r *GoRedis) LPOP(list string) (result interface{}) {
	strCmd := r.Client(list).LPop(r.Ctx, list)
	b, err := strCmd.Bytes()
	if err != nil {
		log.Println("LPOP bytes", err)
		return
	}
	err1 := json.Unmarshal(b, &result)
	if err1 != nil {
		log.Println("LPOP json ", err)
		return
	}
	return
}

//将一个或多个值插入到列表的尾部
func (r *GoRedis) RPUSH(list string, val ...interface{}) bool {
	intCmd := r.Client(list).RPush(r.Ctx, list, val...)
	i, err := intCmd.Result()
	if err != nil {
		log.Println("RPUSH bytes", err)
		return false
	}
	return i == 1
}

//获取列表长度
func (r *GoRedis) LLEN(list string) int64 {
	intCmd := r.Client(list).LLen(r.Ctx, list)
	i, err := intCmd.Result()
	if err != nil {
		log.Println("RPUSH bytes", err)
	}
	return i
}