CL-Softwares
/
modelRT
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(redis hash): fix bug of redis hash 

1.add redis hash init func
2.replace redis model version in go mod
3.add context parameter in redis exec statement

feat(redis set): add new test of RLock and WLock

1.add redis set init func
2.replace redis model version in go mod
3.add context parameter in redis exec statement

fix(logger): add new test of RLock and WLock

1.add compress parameter
2.optimize initLogger function
This commit is contained in:
douxu 2025-03-24 16:37:43 +08:00
parent 25a55b94e8
commit 2f1b9d26b8
12 changed files with 367 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
config/config.go
View File

@ -41,6 +41,16 @@ type LoggerConfig struct {
MaxSize int `mapstructure:"maxsize"`
MaxBackups int `mapstructure:"maxbackups"`
MaxAge int `mapstructure:"maxage"`
Compress bool `mapstructure:"compress"`
}
// RedisConfig define config stuct of redis config
type RedisConfig struct {
Addr string `mapstructure:"addr"`
Password string `mapstructure:"password"`
DB int `mapstructure:"db"`
PoolSize int `mapstructure:"poolsize"`
Timeout int `mapstructure:"timeout"`
}
// AntsConfig define config stuct of ants pool config
@ -65,6 +75,7 @@ type ModelRTConfig struct {
LoggerConfig `mapstructure:"logger"`
AntsConfig `mapstructure:"ants"`
DataRTConfig `mapstructure:"dataRT"`
RedisConfig `mapstructure:"redis"`
PostgresDBURI string `mapstructure:"-"`
}

9
config/config.yaml
View File

@ -29,12 +29,21 @@ logger:
maxsize: 1
maxbackups: 5
maxage: 30
compress: false
# ants config
ants:
parse_concurrent_quantity: 10
rtd_receive_concurrent_quantity: 10
# redis config
redis:
addr: "192.168.2.104:6379"
password: ""
db: 1
poolsize: 50
timeout: 10
# modelRT base config
base:
grid_id: 1

30
diagram/redis_hash.go
View File

@ -3,15 +3,15 @@ package diagram
import (
"context"
distributed_lock "modelRT/distributedlock"
locker "modelRT/distributedlock"
"modelRT/logger"
"github.com/redis/go-redis/v9"
// "github.com/go-redis/redis"
"go.uber.org/zap"
)
// TODO 统一 storageClient与 rwLocker 中使用的 redis 版本
// RedisHash defines the encapsulation struct of redis hash type
type RedisHash struct {
ctx context.Context
@ -20,14 +20,24 @@ type RedisHash struct {
logger *zap.Logger
}
// NewRedisHash define func of new redis hash instance
func NewRedisHash(ctx context.Context, hashKey string, token string, lockLeaseTime uint64, needRefresh bool) *RedisHash {
return &RedisHash{
ctx: ctx,
rwLocker: distributed_lock.InitRWLocker(hashKey, token, lockLeaseTime, needRefresh),
storageClient: GetRedisClientInstance(),
logger: logger.GetLoggerInstance(),
}
}
// SetRedisHashByMap define func of set redis hash by map struct
func (rh *RedisHash) SetRedisHashByMap(hashKey string, fields map[string]interface{}) error {
err := rh.rwLocker.WLock()
err := rh.rwLocker.WLock(rh.ctx)
if err != nil {
rh.logger.Error("lock wLock by hashKey failed", zap.String("hashKey", hashKey), zap.Error(err))
return err
}
defer rh.rwLocker.UnWLock()
defer rh.rwLocker.UnWLock(rh.ctx)
err = rh.storageClient.HSet(rh.ctx, hashKey, fields).Err()
if err != nil {
@ -39,12 +49,12 @@ func (rh *RedisHash) SetRedisHashByMap(hashKey string, fields map[string]interfa
// SetRedisHashByKV define func of set redis hash by kv struct
func (rh *RedisHash) SetRedisHashByKV(hashKey string, field string, value interface{}) error {
err := rh.rwLocker.WLock()
err := rh.rwLocker.WLock(rh.ctx)
if err != nil {
rh.logger.Error("lock wLock by hashKey failed", zap.String("hashKey", hashKey), zap.Error(err))
return err
}
defer rh.rwLocker.UnWLock()
defer rh.rwLocker.UnWLock(rh.ctx)
err = rh.storageClient.HSet(rh.ctx, hashKey, field, value).Err()
if err != nil {
@ -56,12 +66,12 @@ func (rh *RedisHash) SetRedisHashByKV(hashKey string, field string, value interf
// HGet define func of get specified field value from redis hash by key and field name
func (rh *RedisHash) HGet(hashKey string, field string) (string, error) {
err := rh.rwLocker.RLock()
err := rh.rwLocker.RLock(rh.ctx)
if err != nil {
rh.logger.Error("lock rLock by hashKey failed", zap.String("hashKey", hashKey), zap.Error(err))
return "", err
}
defer rh.rwLocker.UnRLock()
defer rh.rwLocker.UnRLock(rh.ctx)
result, err := rh.storageClient.HGet(rh.ctx, hashKey, field).Result()
if err != nil {
@ -73,12 +83,12 @@ func (rh *RedisHash) HGet(hashKey string, field string) (string, error) {
// HGetAll define func of get all filelds from redis hash by key
func (rh *RedisHash) HGetAll(hashKey string) (map[string]string, error) {
err := rh.rwLocker.RLock()
err := rh.rwLocker.RLock(rh.ctx)
if err != nil {
rh.logger.Error("lock rLock by hashKey failed", zap.String("hashKey", hashKey), zap.Error(err))
return nil, err
}
defer rh.rwLocker.UnRLock()
defer rh.rwLocker.UnRLock(rh.ctx)
result, err := rh.storageClient.HGetAll(rh.ctx, hashKey).Result()
if err != nil {

40
diagram/redis_init.go
View File

@ -2,20 +2,44 @@ package diagram
import (
"sync"
"time"
"modelRT/config"
"modelRT/util"
"github.com/redis/go-redis/v9"
)
var (
client *redis.Client
once sync.Once
_globalStorageClient *redis.Client
once sync.Once
)
// GetClientInstance define func of get redis client instance
func GetClientInstance() *redis.Client {
once.Do(func() {
// TODO 根据配置文件初始化 redis client
client = &redis.Client{}
})
// initClient define func of return successfully initialized redis client
func initClient(rCfg config.RedisConfig) *redis.Client {
client, err := util.NewRedisClient(
util.WithAddr(rCfg.Addr),
util.WithPassword(rCfg.Password),
util.WithDB(rCfg.DB),
util.WithPoolSize(rCfg.PoolSize),
util.WithTimeout(time.Duration(rCfg.Timeout)*time.Second),
)
if err != nil {
panic(err)
}
return client
}
// InitClientInstance define func of return instance of redis client
func InitClientInstance(rCfg config.RedisConfig) *redis.Client {
once.Do(func() {
_globalStorageClient = initClient(rCfg)
})
return _globalStorageClient
}
// GetRedisClientInstance define func of get redis client instance
func GetRedisClientInstance() *redis.Client {
client := _globalStorageClient
return client
}

25
diagram/redis_set.go
View File

@ -4,13 +4,14 @@ import (
"context"
"fmt"
distributed_lock "modelRT/distributedlock"
locker "modelRT/distributedlock"
"modelRT/logger"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
)
// TODO 统一 storageClient与 rwLocker 中使用的 redis 版本
// RedisSet defines the encapsulation struct of redis hash type
type RedisSet struct {
ctx context.Context
@ -19,14 +20,24 @@ type RedisSet struct {
logger *zap.Logger
}
// NewRedisSet define func of new redis set instance
func NewRedisSet(ctx context.Context, hashKey string, token string, lockLeaseTime uint64, needRefresh bool) *RedisSet {
return &RedisSet{
ctx: ctx,
rwLocker: distributed_lock.InitRWLocker(hashKey, token, lockLeaseTime, needRefresh),
storageClient: GetRedisClientInstance(),
logger: logger.GetLoggerInstance(),
}
}
// SADD define func of add redis set by members
func (rs *RedisSet) SADD(setKey string, members ...interface{}) error {
err := rs.rwLocker.WLock()
err := rs.rwLocker.WLock(rs.ctx)
if err != nil {
rs.logger.Error("lock wLock by setKey failed", zap.String("setKey", setKey), zap.Error(err))
return err
}
defer rs.rwLocker.UnWLock()
defer rs.rwLocker.UnWLock(rs.ctx)
err = rs.storageClient.SAdd(rs.ctx, setKey, members).Err()
if err != nil {
@ -38,12 +49,12 @@ func (rs *RedisSet) SADD(setKey string, members ...interface{}) error {
// SREM define func of remove the specified members from redis set by key
func (rs *RedisSet) SREM(setKey string, members ...interface{}) error {
err := rs.rwLocker.WLock()
err := rs.rwLocker.WLock(rs.ctx)
if err != nil {
rs.logger.Error("lock wLock by setKey failed", zap.String("setKey", setKey), zap.Error(err))
return err
}
defer rs.rwLocker.UnWLock()
defer rs.rwLocker.UnWLock(rs.ctx)
count, err := rs.storageClient.SRem(rs.ctx, setKey, members).Result()
if err != nil || count != int64(len(members)) {
@ -56,12 +67,12 @@ func (rs *RedisSet) SREM(setKey string, members ...interface{}) error {
// SMembers define func of get all memebers from redis set by key
func (rs *RedisSet) SMembers(setKey string) ([]string, error) {
err := rs.rwLocker.RLock()
err := rs.rwLocker.RLock(rs.ctx)
if err != nil {
rs.logger.Error("lock rLock by setKey failed", zap.String("setKey", setKey), zap.Error(err))
return nil, err
}
defer rs.rwLocker.UnRLock()
defer rs.rwLocker.UnRLock(rs.ctx)
result, err := rs.storageClient.SMembers(rs.ctx, setKey).Result()
if err != nil {

44
distributedlock/locker_client_init.go Normal file
View File

@ -0,0 +1,44 @@
package distributed_lock
import (
"sync"
"time"
"modelRT/config"
"modelRT/util"
"github.com/redis/go-redis/v9"
)
var (
_globalLockerClient *redis.Client
once sync.Once
)
// initClient define func of return successfully initialized redis client
func initClient(rCfg config.RedisConfig) *redis.Client {
client, err := util.NewRedisClient(
util.WithAddr(rCfg.Addr),
util.WithPassword(rCfg.Password),
util.WithPoolSize(rCfg.PoolSize),
util.WithTimeout(time.Duration(rCfg.Timeout)*time.Second),
)
if err != nil {
panic(err)
}
return client
}
// InitClientInstance define func of return instance of redis client
func InitClientInstance(rCfg config.RedisConfig) *redis.Client {
once.Do(func() {
_globalLockerClient = initClient(rCfg)
})
return _globalLockerClient
}
// GetRedisClientInstance define func of get redis client instance
func GetRedisClientInstance() *redis.Client {
client := _globalLockerClient
return client
}

31
distributedlock/redis_lock.go
View File

@ -1,6 +1,7 @@
package distributed_lock
import (
"context"
"errors"
"fmt"
"strings"
@ -11,8 +12,8 @@ import (
luascript "modelRT/distributedlock/luascript"
"modelRT/logger"
"github.com/go-redis/redis"
uuid "github.com/google/uuid"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
)
@ -44,11 +45,11 @@ type redissionLocker struct {
logger *zap.Logger
}
func (rl *redissionLocker) Lock(timeout ...time.Duration) error {
func (rl *redissionLocker) Lock(ctx context.Context, timeout ...time.Duration) error {
if rl.exit == nil {
rl.exit = make(chan struct{})
}
result := rl.tryLock().(*constant.RedisResult)
result := rl.tryLock(ctx).(*constant.RedisResult)
if result.Code == constant.UnknownInternalError {
rl.logger.Error(result.OutputResultMessage())
return fmt.Errorf("get lock failed:%w", result)
@ -57,16 +58,16 @@ func (rl *redissionLocker) Lock(timeout ...time.Duration) error {
if (result.Code == constant.LockSuccess) && rl.needRefresh {
rl.once.Do(func() {
// async refresh lock timeout unitl receive exit singal
go rl.refreshLockTimeout()
go rl.refreshLockTimeout(ctx)
})
return nil
}
subMsg := make(chan struct{}, 1)
defer close(subMsg)
sub := rl.client.Subscribe(rl.waitChanKey)
sub := rl.client.Subscribe(ctx, rl.waitChanKey)
defer sub.Close()
go rl.subscribeLock(sub, subMsg)
go rl.subscribeLock(ctx, sub, subMsg)
if len(timeout) > 0 && timeout[0] > 0 {
acquireTimer := time.NewTimer(timeout[0])
@ -79,7 +80,7 @@ func (rl *redissionLocker) Lock(timeout ...time.Duration) error {
return err
}
resultErr := rl.tryLock().(*constant.RedisResult)
resultErr := rl.tryLock(ctx).(*constant.RedisResult)
if (resultErr.Code == constant.LockFailure) || (resultErr.Code == constant.UnknownInternalError) {
rl.logger.Info(resultErr.OutputResultMessage())
continue
@ -99,14 +100,14 @@ func (rl *redissionLocker) Lock(timeout ...time.Duration) error {
return fmt.Errorf("lock the redis lock failed:%w", result)
}
func (rl *redissionLocker) subscribeLock(sub *redis.PubSub, out chan struct{}) {
func (rl *redissionLocker) subscribeLock(ctx context.Context, sub *redis.PubSub, out chan struct{}) {
if sub == nil || out == nil {
return
}
rl.logger.Info("lock: enter sub routine", zap.String("token", rl.token))
for {
msg, err := sub.Receive()
msg, err := sub.Receive(ctx)
if err != nil {
rl.logger.Info("sub receive message failed", zap.Error(err))
continue
@ -134,7 +135,7 @@ KEYS[1]:锁的键名（key）,通常是锁的唯一标识。
ARGV[1]:锁的过期时间（lockLeaseTime）,单位为秒。
ARGV[2]:当前客户端的唯一标识（token）,用于区分不同的客户端。
*/
func (rl *redissionLocker) refreshLockTimeout() {
func (rl *redissionLocker) refreshLockTimeout(ctx context.Context) {
rl.logger.Info("lock refresh by key and token", zap.String("token", rl.token), zap.String("key", rl.key))
lockTime := time.Duration(rl.lockLeaseTime/3) * time.Second
@ -145,7 +146,7 @@ func (rl *redissionLocker) refreshLockTimeout() {
select {
case <-timer.C:
// extend key lease time
res := rl.client.Eval(luascript.RefreshLockScript, []string{rl.key}, rl.lockLeaseTime, rl.token)
res := rl.client.Eval(ctx, luascript.RefreshLockScript, []string{rl.key}, rl.lockLeaseTime, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
rl.logger.Info("lock refresh failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err))
@ -179,9 +180,9 @@ KEYS[1]:锁的键名（key）,通常是锁的唯一标识。
ARGV[1]:锁的过期时间（lockLeaseTime）,单位为秒。
ARGV[2]:当前客户端的唯一标识（token）,用于区分不同的客户端。
*/
func (rl *redissionLocker) tryLock() error {
func (rl *redissionLocker) tryLock(ctx context.Context) error {
lockType := constant.LockType
res := rl.client.Eval(luascript.LockScript, []string{rl.key}, rl.lockLeaseTime, rl.token)
res := rl.client.Eval(ctx, luascript.LockScript, []string{rl.key}, rl.lockLeaseTime, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
return constant.NewRedisResult(constant.UnknownInternalError, lockType, err.Error())
@ -195,8 +196,8 @@ KEYS[2]:锁的释放通知频道（chankey）,用于通知其他客户端锁已
ARGV[1]:解锁消息（unlockMessage）,用于通知其他客户端锁已释放。
ARGV[2]:当前客户端的唯一标识（token）,用于区分不同的客户端。
*/
func (rl *redissionLocker) UnLock() error {
res := rl.client.Eval(luascript.UnLockScript, []string{rl.key, rl.waitChanKey}, unlockMessage, rl.token)
func (rl *redissionLocker) UnLock(ctx context.Context) error {
res := rl.client.Eval(ctx, luascript.UnLockScript, []string{rl.key, rl.waitChanKey}, unlockMessage, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
rl.logger.Info("unlock lock failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err))

58
distributedlock/redis_rwlock.go
View File

@ -1,6 +1,7 @@
package distributed_lock
import (
"context"
"errors"
"fmt"
"strings"
@ -11,8 +12,8 @@ import (
"modelRT/distributedlock/luascript"
"modelRT/logger"
"github.com/go-redis/redis"
uuid "github.com/google/uuid"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
)
@ -21,12 +22,12 @@ type RedissionRWLocker struct {
rwTokenTimeoutPrefix string
}
func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error {
func (rl *RedissionRWLocker) RLock(ctx context.Context, timeout ...time.Duration) error {
if rl.exit == nil {
rl.exit = make(chan struct{})
}
result := rl.tryRLock().(*constant.RedisResult)
result := rl.tryRLock(ctx).(*constant.RedisResult)
if result.Code == constant.UnknownInternalError {
rl.logger.Error(result.OutputResultMessage())
return fmt.Errorf("get read lock failed:%w", result)
@ -36,7 +37,7 @@ func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error {
if rl.needRefresh {
rl.once.Do(func() {
// async refresh lock timeout unitl receive exit singal
go rl.refreshLockTimeout()
go rl.refreshLockTimeout(ctx)
})
}
rl.logger.Info("success get the read by key and token", zap.String("key", rl.key), zap.String("token", rl.token))
@ -45,9 +46,9 @@ func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error {
subMsg := make(chan struct{}, 1)
defer close(subMsg)
sub := rl.client.Subscribe(rl.waitChanKey)
sub := rl.client.Subscribe(ctx, rl.waitChanKey)
defer sub.Close()
go rl.subscribeLock(sub, subMsg)
go rl.subscribeLock(ctx, sub, subMsg)
if len(timeout) > 0 && timeout[0] > 0 {
acquireTimer := time.NewTimer(timeout[0])
@ -60,7 +61,7 @@ func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error {
return err
}
resultErr := rl.tryRLock().(*constant.RedisResult)
resultErr := rl.tryRLock(ctx).(*constant.RedisResult)
if (resultErr.Code == constant.RLockFailureWithWLockOccupancy) || (resultErr.Code == constant.UnknownInternalError) {
rl.logger.Info(resultErr.OutputResultMessage())
continue
@ -80,10 +81,10 @@ func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error {
return fmt.Errorf("lock the redis read lock failed:%w", result)
}
func (rl *RedissionRWLocker) tryRLock() error {
func (rl *RedissionRWLocker) tryRLock(ctx context.Context) error {
lockType := constant.LockType
res := rl.client.Eval(luascript.RLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
res := rl.client.Eval(ctx, luascript.RLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
return constant.NewRedisResult(constant.UnknownInternalError, lockType, err.Error())
@ -91,7 +92,7 @@ func (rl *RedissionRWLocker) tryRLock() error {
return constant.NewRedisResult(constant.RedisCode(val), lockType, "")
}
func (rl *RedissionRWLocker) refreshLockTimeout() {
func (rl *RedissionRWLocker) refreshLockTimeout(ctx context.Context) {
rl.logger.Info("lock refresh by key and token", zap.String("token", rl.token), zap.String("key", rl.key))
lockTime := time.Duration(rl.lockLeaseTime/3) * time.Second
@ -102,7 +103,7 @@ func (rl *RedissionRWLocker) refreshLockTimeout() {
select {
case <-timer.C:
// extend key lease time
res := rl.client.Eval(luascript.RefreshRWLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
res := rl.client.Eval(ctx, luascript.RefreshRWLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
rl.logger.Info("lock refresh failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err))
@ -124,9 +125,9 @@ func (rl *RedissionRWLocker) refreshLockTimeout() {
}
}
func (rl *RedissionRWLocker) UnRLock() error {
func (rl *RedissionRWLocker) UnRLock(ctx context.Context) error {
rl.logger.Info("unlock RLock by key and token", zap.String("key", rl.key), zap.String("token", rl.token))
res := rl.client.Eval(luascript.UnRLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix, rl.waitChanKey}, unlockMessage, rl.token)
res := rl.client.Eval(ctx, luascript.UnRLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix, rl.waitChanKey}, unlockMessage, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
rl.logger.Info("unlock read lock failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err))
@ -149,12 +150,12 @@ func (rl *RedissionRWLocker) UnRLock() error {
return nil
}
func (rl *RedissionRWLocker) WLock(timeout ...time.Duration) error {
func (rl *RedissionRWLocker) WLock(ctx context.Context, timeout ...time.Duration) error {
if rl.exit == nil {
rl.exit = make(chan struct{})
}
result := rl.tryWLock().(*constant.RedisResult)
result := rl.tryWLock(ctx).(*constant.RedisResult)
if result.Code == constant.UnknownInternalError {
rl.logger.Error(result.OutputResultMessage())
return fmt.Errorf("get write lock failed:%w", result)
@ -163,16 +164,16 @@ func (rl *RedissionRWLocker) WLock(timeout ...time.Duration) error {
if (result.Code == constant.LockSuccess) && rl.needRefresh {
rl.once.Do(func() {
// async refresh lock timeout unitl receive exit singal
go rl.refreshLockTimeout()
go rl.refreshLockTimeout(ctx)
})
return nil
}
subMsg := make(chan struct{}, 1)
defer close(subMsg)
sub := rl.client.Subscribe(rl.waitChanKey)
sub := rl.client.Subscribe(ctx, rl.waitChanKey)
defer sub.Close()
go rl.subscribeLock(sub, subMsg)
go rl.subscribeLock(ctx, sub, subMsg)
if len(timeout) > 0 && timeout[0] > 0 {
acquireTimer := time.NewTimer(timeout[0])
@ -185,7 +186,7 @@ func (rl *RedissionRWLocker) WLock(timeout ...time.Duration) error {
return err
}
result := rl.tryWLock().(*constant.RedisResult)
result := rl.tryWLock(ctx).(*constant.RedisResult)
if (result.Code == constant.UnknownInternalError) || (result.Code == constant.WLockFailureWithRLockOccupancy) || (result.Code == constant.WLockFailureWithWLockOccupancy) || (result.Code == constant.WLockFailureWithNotFirstPriority) {
rl.logger.Info(result.OutputResultMessage())
continue
@ -205,10 +206,10 @@ func (rl *RedissionRWLocker) WLock(timeout ...time.Duration) error {
return fmt.Errorf("lock write lock failed:%w", result)
}
func (rl *RedissionRWLocker) tryWLock() error {
func (rl *RedissionRWLocker) tryWLock(ctx context.Context) error {
lockType := constant.LockType
res := rl.client.Eval(luascript.WLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
res := rl.client.Eval(ctx, luascript.WLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix}, rl.lockLeaseTime, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
return constant.NewRedisResult(constant.UnknownInternalError, lockType, err.Error())
@ -216,8 +217,8 @@ func (rl *RedissionRWLocker) tryWLock() error {
return constant.NewRedisResult(constant.RedisCode(val), lockType, "")
}
func (rl *RedissionRWLocker) UnWLock() error {
res := rl.client.Eval(luascript.UnWLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix, rl.waitChanKey}, unlockMessage, rl.token)
func (rl *RedissionRWLocker) UnWLock(ctx context.Context) error {
res := rl.client.Eval(ctx, luascript.UnWLockScript, []string{rl.key, rl.rwTokenTimeoutPrefix, rl.waitChanKey}, unlockMessage, rl.token)
val, err := res.Int()
if err != redis.Nil && err != nil {
rl.logger.Error("unlock write lock failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err))
@ -278,3 +279,14 @@ func GetRWLocker(client *redis.Client, ops *RedissionLockConfig) *RedissionRWLoc
}
return rwLocker
}
// TODO consider refactoring to use options mode
func InitRWLocker(key string, token string, lockLeaseTime uint64, needRefresh bool) *RedissionRWLocker {
ops := &RedissionLockConfig{
Key: key,
Token: token,
LockLeaseTime: lockLeaseTime,
NeedRefresh: needRefresh,
}
return GetRWLocker(GetRedisClientInstance(), ops)
}

108
distributedlock/rwlock_test.go
View File

@ -1,11 +1,12 @@
package distributed_lock
import (
"context"
"strings"
"testing"
"time"
"github.com/go-redis/redis"
"github.com/redis/go-redis/v9"
"github.com/stretchr/testify/assert"
"go.uber.org/zap"
)
@ -17,6 +18,7 @@ func init() {
}
func TestRWLockRLockAndUnRLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -35,18 +37,18 @@ func TestRWLockRLockAndUnRLock(t *testing.T) {
duration := 10 * time.Second
// 第一次加读锁
err := rwLocker.RLock(duration)
err := rwLocker.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey := strings.Join([]string{rwLocker.rwTokenTimeoutPrefix, rwLocker.token}, ":")
num, err := rdb.HGet(rwLocker.key, tokenKey).Int()
num, err := rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
err = rwLocker.UnRLock()
err = rwLocker.UnRLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, redis.Nil, err)
assert.Equal(t, 0, num)
t.Log("rwLock rlock and unrlock test success")
@ -54,6 +56,7 @@ func TestRWLockRLockAndUnRLock(t *testing.T) {
}
func TestRWLockReentrantRLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -72,35 +75,35 @@ func TestRWLockReentrantRLock(t *testing.T) {
duration := 10 * time.Second
// 第一次加读锁
err := rwLocker.RLock(duration)
err := rwLocker.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey := strings.Join([]string{rwLocker.rwTokenTimeoutPrefix, rwLocker.token}, ":")
num, err := rdb.HGet(rwLocker.key, tokenKey).Int()
num, err := rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// 第二次加读锁
err = rwLocker.RLock(duration)
err = rwLocker.RLock(ctx, duration)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 2, num)
// 第一次解读锁
err = rwLocker.UnRLock()
err = rwLocker.UnRLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// 第二次解读锁
err = rwLocker.UnRLock()
err = rwLocker.UnRLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, redis.Nil, err)
assert.Equal(t, 0, num)
t.Log("rwLock reentrant lock test success")
@ -108,6 +111,7 @@ func TestRWLockReentrantRLock(t *testing.T) {
}
func TestRWLockRefreshRLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -126,17 +130,17 @@ func TestRWLockRefreshRLock(t *testing.T) {
duration := 10 * time.Second
// 第一次加读锁
err := rwLocker.RLock(duration)
err := rwLocker.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey := strings.Join([]string{rwLocker.rwTokenTimeoutPrefix, rwLocker.token}, ":")
num, err := rdb.HGet(rwLocker.key, tokenKey).Int()
num, err := rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
time.Sleep(10 * time.Second)
script := `return redis.call('httl', KEYS[1], 'fields', '1', ARGV[1]);`
result, err := rdb.Eval(script, []string{rwLocker.key}, tokenKey).Result()
result, err := rdb.Eval(ctx, script, []string{rwLocker.key}, tokenKey).Result()
assert.Equal(t, nil, err)
ttls, ok := result.([]interface{})
assert.Equal(t, true, ok)
@ -145,10 +149,10 @@ func TestRWLockRefreshRLock(t *testing.T) {
compareValue := int64(8)
assert.Greater(t, ttl, compareValue)
err = rwLocker.UnRLock()
err = rwLocker.UnRLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, redis.Nil, err)
assert.Equal(t, 0, num)
t.Log("rwLock refresh lock test success")
@ -157,6 +161,7 @@ func TestRWLockRefreshRLock(t *testing.T) {
// TODO 设计两个客户端分别加读锁,测试是否可以加锁成功
func TestRWLock2ClientRLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -183,39 +188,39 @@ func TestRWLock2ClientRLock(t *testing.T) {
duration := 10 * time.Second
// locker1加读锁
err := rwLocker1.RLock(duration)
err := rwLocker1.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey1 := strings.Join([]string{rwLocker1.rwTokenTimeoutPrefix, rwLocker1.token}, ":")
num, err := rdb.HGet(rwLocker1.key, tokenKey1).Int()
num, err := rdb.HGet(ctx, rwLocker1.key, tokenKey1).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// locker2加读锁
err = rwLocker2.RLock(duration)
err = rwLocker2.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey2 := strings.Join([]string{rwLocker2.rwTokenTimeoutPrefix, rwLocker2.token}, ":")
num, err = rdb.HGet(rwLocker2.key, tokenKey2).Int()
num, err = rdb.HGet(ctx, rwLocker2.key, tokenKey2).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
err = rdb.HLen(rwLocker1.key).Err()
err = rdb.HLen(ctx, rwLocker1.key).Err()
assert.Equal(t, nil, err)
hLen := rdb.HLen(rwLocker1.key).Val()
hLen := rdb.HLen(ctx, rwLocker1.key).Val()
assert.Equal(t, 3, hLen)
// locker1解读锁
err = rwLocker1.UnRLock()
err = rwLocker1.UnRLock(ctx)
assert.Equal(t, nil, err)
// locker1解读锁
err = rwLocker2.UnRLock()
err = rwLocker2.UnRLock(ctx)
assert.Equal(t, nil, err)
err = rdb.Exists(rwLocker1.key).Err()
err = rdb.Exists(ctx, rwLocker1.key).Err()
assert.Equal(t, redis.Nil, err)
existNum := rdb.Exists(rwLocker1.key).Val()
existNum := rdb.Exists(ctx, rwLocker1.key).Val()
assert.Equal(t, 0, existNum)
t.Log("rwLock 2 client lock test success")
return
@ -223,6 +228,7 @@ func TestRWLock2ClientRLock(t *testing.T) {
// TODO 设计两个客户端分别加时间不同的读锁,测试ttl时间在有一个key删除后是否可以变换成功
func TestRWLock2CWith2DifTimeRLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -249,45 +255,46 @@ func TestRWLock2CWith2DifTimeRLock(t *testing.T) {
duration := 10 * time.Second
// locker1加读锁
err := rwLocker1.RLock(duration)
err := rwLocker1.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey1 := strings.Join([]string{rwLocker1.rwTokenTimeoutPrefix, rwLocker1.token}, ":")
num, err := rdb.HGet(rwLocker1.key, tokenKey1).Int()
num, err := rdb.HGet(ctx, rwLocker1.key, tokenKey1).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// locker2加读锁
err = rwLocker2.RLock(duration)
err = rwLocker2.RLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey2 := strings.Join([]string{rwLocker2.rwTokenTimeoutPrefix, rwLocker2.token}, ":")
num, err = rdb.HGet(rwLocker2.key, tokenKey2).Int()
num, err = rdb.HGet(ctx, rwLocker2.key, tokenKey2).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
err = rdb.HLen(rwLocker1.key).Err()
err = rdb.HLen(ctx, rwLocker1.key).Err()
assert.Equal(t, nil, err)
hLen := rdb.HLen(rwLocker1.key).Val()
hLen := rdb.HLen(ctx, rwLocker1.key).Val()
assert.Equal(t, 3, hLen)
// locker1解读锁
err = rwLocker1.UnRLock()
err = rwLocker1.UnRLock(ctx)
assert.Equal(t, nil, err)
// locker1解读锁
err = rwLocker2.UnRLock()
err = rwLocker2.UnRLock(ctx)
assert.Equal(t, nil, err)
err = rdb.Exists(rwLocker1.key).Err()
err = rdb.Exists(ctx, rwLocker1.key).Err()
assert.Equal(t, redis.Nil, err)
existNum := rdb.Exists(rwLocker1.key).Val()
existNum := rdb.Exists(ctx, rwLocker1.key).Val()
assert.Equal(t, 0, existNum)
t.Log("rwLock 2 client lock test success")
return
}
func TestRWLockWLockAndUnWLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -306,18 +313,18 @@ func TestRWLockWLockAndUnWLock(t *testing.T) {
duration := 10 * time.Second
// 第一次加读锁
err := rwLocker.WLock(duration)
err := rwLocker.WLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey := strings.Join([]string{rwLocker.rwTokenTimeoutPrefix, rwLocker.token}, ":")
num, err := rdb.HGet(rwLocker.key, tokenKey).Int()
num, err := rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
err = rwLocker.UnWLock()
err = rwLocker.UnWLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, redis.Nil, err)
assert.Equal(t, 0, num)
t.Log("rwLock rlock and unrlock test success")
@ -325,6 +332,7 @@ func TestRWLockWLockAndUnWLock(t *testing.T) {
}
func TestRWLockReentrantWLock(t *testing.T) {
ctx := context.TODO()
rdb := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: "192.168.2.104:6379",
@ -343,35 +351,35 @@ func TestRWLockReentrantWLock(t *testing.T) {
duration := 10 * time.Second
// 第一次加写锁
err := rwLocker.WLock(duration)
err := rwLocker.WLock(ctx, duration)
assert.Equal(t, nil, err)
tokenKey := strings.Join([]string{rwLocker.rwTokenTimeoutPrefix, rwLocker.token}, ":")
num, err := rdb.HGet(rwLocker.key, tokenKey).Int()
num, err := rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// 第二次加写锁
err = rwLocker.WLock(duration)
err = rwLocker.WLock(ctx, duration)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 2, num)
// 第一次解写锁
err = rwLocker.UnWLock()
err = rwLocker.UnWLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, nil, err)
assert.Equal(t, 1, num)
// 第二次解写锁
err = rwLocker.UnWLock()
err = rwLocker.UnWLock(ctx)
assert.Equal(t, nil, err)
num, err = rdb.HGet(rwLocker.key, tokenKey).Int()
num, err = rdb.HGet(ctx, rwLocker.key, tokenKey).Int()
assert.Equal(t, redis.Nil, err)
assert.Equal(t, 0, num)
t.Log("rwLock reentrant lock test success")

13
logger/init.go
View File

@ -31,13 +31,13 @@ func getEncoder() zapcore.Encoder {
}
// getLogWriter responsible for setting the location of log storage
func getLogWriter(mode, filename string, maxsize, maxBackup, maxAge int) zapcore.WriteSyncer {
func getLogWriter(mode, filename string, maxsize, maxBackup, maxAge int, compress bool) zapcore.WriteSyncer {
lumberJackLogger := &lumberjack.Logger{
Filename: filename, // log file position
MaxSize: maxsize, // log file maxsize
MaxAge: maxAge, // maximum number of day files retained
MaxBackups: maxBackup, // maximum number of old files retained
Compress: false, // whether to compress
Compress: compress, // whether to compress
}
syncConsole := zapcore.AddSync(os.Stderr)
@ -51,7 +51,7 @@ func getLogWriter(mode, filename string, maxsize, maxBackup, maxAge int) zapcore
// initLogger return successfully initialized zap logger
func initLogger(lCfg config.LoggerConfig) *zap.Logger {
writeSyncer := getLogWriter(lCfg.Mode, lCfg.FilePath, lCfg.MaxSize, lCfg.MaxBackups, lCfg.MaxAge)
writeSyncer := getLogWriter(lCfg.Mode, lCfg.FilePath, lCfg.MaxSize, lCfg.MaxBackups, lCfg.MaxAge, lCfg.Compress)
encoder := getEncoder()
l := new(zapcore.Level)
@ -61,10 +61,11 @@ func initLogger(lCfg config.LoggerConfig) *zap.Logger {
}
core := zapcore.NewCore(encoder, writeSyncer, l)
_globalLogger = zap.New(core, zap.AddCaller())
zap.ReplaceGlobals(_globalLogger)
logger := zap.New(core, zap.AddCaller())
return _globalLogger
// 替换全局日志实例
zap.ReplaceGlobals(logger)
return logger
}
// InitLoggerInstance return instance of zap logger

8
main.go
View File

@ -9,6 +9,8 @@ import (
"modelRT/alert"
"modelRT/config"
"modelRT/database"
"modelRT/diagram"
distributed_lock "modelRT/distributedlock"
_ "modelRT/docs"
"modelRT/handler"
"modelRT/logger"
@ -77,6 +79,12 @@ func main() {
}
defer parsePool.Release()
storageClient := diagram.InitClientInstance(modelRTConfig.RedisConfig)
defer storageClient.Close()
lockerClient := distributed_lock.InitClientInstance(modelRTConfig.RedisConfig)
defer lockerClient.Close()
// init anchor param ants pool
anchorRealTimePool, err := pool.AnchorPoolInit(modelRTConfig.RTDReceiveConcurrentQuantity)
if err != nil {

109
util/redis_options.go Normal file
View File

@ -0,0 +1,109 @@
package util
import (
"context"
"errors"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
// RedisOptions define struct of redis client config options
type RedisOptions struct {
Addr string
Password string
DB int
PoolSize int
Timeout time.Duration
}
// RedisOption define a function type for modify RedisOptions
type RedisOption func(*RedisOptions) error
// WithAddr define func of configure redis addr options
func WithAddr(addr string) RedisOption {
return func(o *RedisOptions) error {
if addr == "" {
return errors.New("地址不能为空")
}
o.Addr = addr
return nil
}
}
// WithPassword define func of configure redis password options
func WithPassword(password string) RedisOption {
return func(o *RedisOptions) error {
o.Password = password
return nil
}
}
// WithDB define func of configure redis db options
func WithDB(db int) RedisOption {
return func(o *RedisOptions) error {
if db < 0 {
return errors.New("数据库编号不能为负数")
}
o.DB = db
return nil
}
}
// WithPoolSize define func of configure pool size options
func WithPoolSize(poolSize int) RedisOption {
return func(o *RedisOptions) error {
if poolSize <= 0 {
return errors.New("连接池大小必须大于 0")
}
o.PoolSize = poolSize
return nil
}
}
// WithTimeout define func of configure timeout options
func WithTimeout(timeout time.Duration) RedisOption {
return func(o *RedisOptions) error {
if timeout <= 0 {
return errors.New("超时时间必须大于 0")
}
o.Timeout = timeout
return nil
}
}
// NewRedisClient define func of initialize the Redis client
func NewRedisClient(opts ...RedisOption) (*redis.Client, error) {
// default options
options := &RedisOptions{
Addr: "localhost:6379",
Password: "",
DB: 0,
PoolSize: 10,
Timeout: 5 * time.Second,
}
// Apply configuration options from config
for _, opt := range opts {
if err := opt(options); err != nil {
return nil, err
}
}
// create redis client
client := redis.NewClient(&redis.Options{
Addr: options.Addr,
Password: options.Password,
DB: options.DB,
PoolSize: options.PoolSize,
})
// check if the connection is successful
ctx, cancel := context.WithTimeout(context.Background(), options.Timeout)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
return nil, fmt.Errorf("can not connect redis:%v", err)
}
return client, nil
}