redis/client.go

440 lines
11 KiB
Go

// Package redis ...
//
// Description : redis 客户端
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 2021-02-27 4:49 下午
package redis
import (
"encoding/json"
"fmt"
"git.zhangdeman.cn/zhangdeman/consts"
"git.zhangdeman.cn/zhangdeman/serialize"
"path/filepath"
"strings"
"sync"
"time"
"git.zhangdeman.cn/zhangdeman/logger"
redisInstance "github.com/go-redis/redis/v8"
"go.uber.org/zap"
)
var (
// Client 连接实例
Client ClientInterface
)
// InitWithCfgFile 使用配置文件初始化
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 21:24 2022/6/22
func InitWithCfgFile(cfgFilePath string) error {
c := &OwnClient{
lock: &sync.RWMutex{},
instanceTable: make(map[string]*RealClient),
confTable: make(map[string]*FullConfig),
parseErrorFunc: defaultParseError,
}
if err := c.loadConfig(cfgFilePath); nil != err {
return err
}
if err := c.init(); nil != err {
return err
}
Client = c
return nil
}
// InitWithCfgDir 使用配置目录初始化
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 21:31 2022/6/22
func InitWithCfgDir(cfgDir string) error {
c := &OwnClient{
lock: &sync.RWMutex{},
instanceTable: make(map[string]*RealClient),
confTable: make(map[string]*FullConfig),
parseErrorFunc: defaultParseError,
}
if err := c.batchLoadConfig(cfgDir); nil != err {
return err
}
if err := c.init(); nil != err {
return err
}
Client = c
return nil
}
// InitWithCfgFileList 使用批量的配置文件初始化
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 21:33 2022/6/22
func InitWithCfgFileList(cfgFileList []string) error {
c := &OwnClient{
lock: &sync.RWMutex{},
instanceTable: make(map[string]*RealClient),
confTable: make(map[string]*FullConfig),
parseErrorFunc: defaultParseError,
}
for _, cfgFilePath := range cfgFileList {
if err := c.loadConfig(cfgFilePath); nil != err {
return err
}
}
if err := c.init(); nil != err {
return err
}
Client = c
return nil
}
// defaultParseError ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 10:59 下午 2021/2/27
func defaultParseError(err error) error {
if nil == err {
return nil
}
errMsg := err.Error()
if errMsg == "nil" || errMsg == "<nil>" {
return nil
}
strArr := strings.Split(errMsg, ":")
if len(strArr) != 2 {
return err
}
msg := strings.ToLower(strings.TrimSpace(strArr[1]))
if msg == "nil" || msg == "<nil>" {
return nil
}
return err
}
// RealClient 包装好的 redis client
type RealClient struct {
Flag string // redis 标识
Master *redisInstance.Client // redis 实例
Slave *redisInstance.Client // redis 实例
Logger *zap.Logger // 日志实例
}
// NewClient 获取redis client实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 5:05 下午 2021/2/27
func NewClient(config map[string]*FullConfig, parseErrorFunc func(err error) error) (ClientInterface, error) {
c := &OwnClient{
lock: &sync.RWMutex{},
instanceTable: make(map[string]*RealClient),
loggerTable: make(map[string]*zap.Logger),
confTable: config,
parseErrorFunc: parseErrorFunc,
}
if nil == c.parseErrorFunc {
c.parseErrorFunc = defaultParseError
}
return c, c.init()
}
// OwnClient 包装的redis client
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 4:52 下午 2021/2/27
type OwnClient struct {
lock *sync.RWMutex // 锁
loggerTable map[string]*zap.Logger // 日志实例
instanceTable map[string]*RealClient // redis 实例
confTable map[string]*FullConfig // redis 配置
parseErrorFunc func(err error) error // 解析err的function,解析执行结果是否为失败,有的场景,执行成功,返回 redis:nil / redis:<nil>
}
// AddClientWithCfgFile 使用具体配置文件初始化
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 20:57 2022/6/15
func (c *OwnClient) AddClientWithCfgFile(cfgPath string) error {
var (
err error
)
if err = c.loadConfig(cfgPath); nil != err {
return err
}
return c.init()
}
// AddClientWithCfgDir 使用配置目录进行初始化
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 20:58 2022/6/15
func (c *OwnClient) AddClientWithCfgDir(cfgDir string) error {
var (
err error
)
if err = c.batchLoadConfig(cfgDir); nil != err {
return err
}
return c.init()
}
// AddClientWithCfg 通过配置增加实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 11:45 2023/8/16
func (c *OwnClient) AddClientWithCfg(flag string, cfg *FullConfig) error {
c.lock.Lock()
c.confTable[flag] = cfg
c.lock.Unlock()
return c.init()
}
// loadConfig 载入配置文件
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 14:31 2022/6/15
func (c *OwnClient) loadConfig(cfgPath string) error {
var (
err error
cfg FullConfig
)
filePathArr := strings.Split(cfgPath, string(filepath.Separator))
if len(filePathArr) == 0 {
return CfgFilePathError()
}
fileName := strings.ToLower(filePathArr[len(filePathArr)-1])
fileArr := strings.Split(fileName, ".")
if len(filePathArr) < 2 {
return CfgFileFormatErr("未知")
}
flag := strings.Trim(
strings.Trim(
strings.Trim(fileName, ".json"),
".yaml"), ".yml")
switch strings.ToLower(fileArr[len(fileArr)-1]) {
case consts.FileTypeJson:
if err = serialize.File.ReadJSONContent(cfgPath, &cfg); nil != err {
return err
}
flag = strings.Join(fileArr[0:len(fileArr)-1], ".")
case consts.FileTypeYml:
fallthrough
case consts.FileTypeYaml:
if err = serialize.File.ReadYmlContent(cfgPath, &cfg); nil != err {
return err
}
flag = strings.Join(fileArr[0:len(fileArr)-1], ".")
default:
// 不支持的格式,跳过
}
c.lock.Lock()
c.confTable[flag] = &cfg
c.lock.Unlock()
return nil
}
// batchLoadConfig 批量载入配置
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 16:44 2022/6/15
func (c *OwnClient) batchLoadConfig(cfgDir string) error {
filepathNames, _ := filepath.Glob(filepath.Join(cfgDir, "*"))
for i := range filepathNames {
if err := c.loadConfig(filepathNames[i]); nil != err {
return err
}
}
return nil
}
// init 初始化redis连接
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 5:31 下午 2021/2/27
func (c *OwnClient) init() error {
var err error
for flag, conf := range c.confTable {
if _, exist := c.instanceTable[flag]; exist {
// 实例已经初始化
continue
}
c.instanceTable[flag] = &RealClient{
Flag: flag,
Master: redisInstance.NewClient(Config2Options(conf.Master)),
Slave: redisInstance.NewClient(Config2Options(conf.Slave)),
}
if c.instanceTable[flag].Logger, err = logger.GetLogInstanceFromInputConfig(conf.Logger); nil != err {
return InitLoggerErr(flag, err)
}
}
return nil
}
// initLogger 初始化日志
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 7:07 下午 2021/2/27
func (c *OwnClient) initLogger(flag string, conf *logger.InputLogConfig) error {
var err error
c.loggerTable[flag], err = logger.GetLogInstanceFromInputConfig(conf)
return LoggerInitFail(flag, err)
}
// GetRedisClientWithError 获取redis实例
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 5:16 下午 2021/2/27
func (c *OwnClient) GetRedisClientWithError(flag string) (*RealClient, error) {
c.lock.RLock()
defer c.lock.RUnlock()
redisClient, exist := c.instanceTable[flag]
if !exist {
return nil, FlagNotFound(flag)
}
return redisClient, nil
}
// GetRedisClient ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 11:52 2023/8/16
func (c *OwnClient) GetRedisClient(flag string) *RealClient {
redisClient, _ := c.GetRedisClientWithError(flag)
return redisClient
}
// log 记录redis请求日志
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 8:52 下午 2021/2/27
func (c *OwnClient) log(ctx *Context, realClient *RealClient, isMaster bool, cmdResult redisInstance.Cmder, startTime int64, finishTime int64) {
if nil == realClient || nil == realClient.Logger {
return
}
realClient.Logger.Info(
"执行redis命令日志记录",
zap.Any(ctx.RequestIDField, ctx.RequestID), // 上下文串联的requestID
zap.String("instance_flag", ctx.Flag), // redis 实例的标识
zap.Any("exec_command_name", cmdResult.Name()), // 执行的命令
zap.Any("exec_command_arg", cmdResult.Args()), // 执行的命令参数
zap.Bool("use_master", isMaster), // 是否使用主节点
zap.Float64("exec_used_time", float64(finishTime-startTime)/1e6), // 耗时,单位: ms
zap.Error(cmdResult.Err()), // 异常信息
zap.String("command_result", cmdResult.String()), // 命令执行后的结果
)
}
// CommandProxy 执行命令的代理
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 9:41 下午 2021/2/27
func (c *OwnClient) CommandProxy(ctx *Context, flag string, cmd string, param ...interface{}) (string, error) {
var (
realClient *RealClient
err error
)
if len(cmd) == 0 {
return "", EmptyCmd()
}
if nil == ctx {
ctx = NewContext(flag)
}
if realClient, err = c.GetRedisClientWithError(ctx.Flag); nil != err {
return "", err
}
flagArr := strings.Split(flag, "#")
if len(flagArr) == 1 {
flagArr = append(flagArr, "r")
}
isMater := flagArr[1] == "w"
redisClient := realClient.Slave
if isMater {
redisClient = realClient.Master
}
redisCmd := append([]interface{}{cmd}, param...)
startTime := time.Now().Unix()
cmdResult := redisClient.Do(ctx.Ctx, redisCmd...)
go c.log(ctx, realClient, isMater, cmdResult, startTime, time.Now().UnixNano())
return fmt.Sprintf("%v", cmdResult.Val()), c.parseErrorFunc(cmdResult.Err())
}
// CommandProxyWithReceiver 执行命令,并解析结果
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 10:00 下午 2021/2/27
func (c *OwnClient) CommandProxyWithReceiver(ctx *Context, flag string, receiver interface{}, cmd string, param ...interface{}) error {
if nil == receiver {
return ReceiverISNIL()
}
var (
err error
result string
)
if result, err = c.CommandProxy(ctx, flag, cmd, param...); nil != err {
return err
}
return ResultConvertFail(json.Unmarshal([]byte(result), receiver))
}
// RemoveClient 移除client
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 11:49 2023/8/16
func (c *OwnClient) RemoveClient(flag string) {
c.lock.Lock()
defer c.lock.Unlock()
delete(c.confTable, flag)
if _, exist := c.instanceTable[flag]; exist {
if nil != c.instanceTable[flag].Master {
_ = c.instanceTable[flag].Master.Close()
}
if nil != c.instanceTable[flag].Slave {
_ = c.instanceTable[flag].Slave.Close()
}
}
delete(c.instanceTable, flag)
}
// ClientInterface 定义redis client的接口实现,方便单元测试数据mock
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 10:49 下午 2021/2/27
type ClientInterface interface {
GetRedisClient(flag string) *RealClient
GetRedisClientWithError(flag string) (*RealClient, error)
CommandProxy(ctx *Context, flag string, cmd string, param ...interface{}) (string, error)
CommandProxyWithReceiver(ctx *Context, flag string, receiver interface{}, cmd string, param ...interface{}) error
AddClientWithCfgFile(cfgPath string) error
AddClientWithCfgDir(cfgDir string) error
AddClientWithCfg(flag string, cfg *FullConfig) error
RemoveClient(flag string)
}