package distributed_lock import ( "errors" "fmt" "strings" "time" "modelRT/distributedlock/constant" "modelRT/distributedlock/luascript" "modelRT/logger" "github.com/go-redis/redis" uuid "github.com/google/uuid" "go.uber.org/zap" ) type RedissionRWLocker struct { redissionLocker writeWaitChanKey string rwTimeoutPrefix string needRefresh bool } func (rl *RedissionRWLocker) RLock(timeout ...time.Duration) error { if rl.exit == nil { rl.exit = make(chan struct{}) } result := rl.tryRLock().(*constant.RedisResult) if result.Code == constant.UnknownInternalError { rl.logger.Error(result.OutputResultMessage()) return fmt.Errorf("get read lock failed:%w", result) } if (result.Code == constant.LockSuccess) && rl.needRefresh { rl.once.Do(func() { // async refresh lock timeout unitl receive exit singal go rl.refreshLockTimeout() }) return nil } subMsg := make(chan struct{}, 1) defer close(subMsg) sub := rl.client.Subscribe(rl.writeWaitChanKey) defer sub.Close() go rl.subscribeLock(sub, subMsg) if len(timeout) > 0 && timeout[0] > 0 { acquireTimer := time.NewTimer(timeout[0]) for { select { case _, ok := <-subMsg: if !ok { err := errors.New("failed to read the read lock waiting for for the channel message") rl.logger.Error("failed to read the read lock waiting for for the channel message") return err } resultErr := rl.tryRLock().(*constant.RedisResult) if (resultErr.Code == constant.RLockFailureWithWLockOccupancy) || (resultErr.Code == constant.UnknownInternalError) { rl.logger.Info(resultErr.OutputResultMessage()) continue } if resultErr.Code == constant.LockSuccess { rl.logger.Info(resultErr.OutputResultMessage()) return nil } case <-acquireTimer.C: err := errors.New("the waiting time for obtaining the read lock operation has timed out") rl.logger.Info("the waiting time for obtaining the read lock operation has timed out") return err } } } return fmt.Errorf("lock read lock failed:%w", result) } func (rl *RedissionRWLocker) tryRLock() error { lockType := constant.LockType res := rl.client.Eval(luascript.RLockScript, []string{rl.key, rl.rwTimeoutPrefix}, rl.lockLeaseTime, rl.token) val, err := res.Int() if err != redis.Nil && err != nil { return constant.NewRedisResult(constant.UnknownInternalError, lockType, err.Error()) } return constant.NewRedisResult(constant.RedisCode(val), lockType, "") } func (rl *RedissionRWLocker) refreshLockTimeout() { rl.logger.Info("read lock refresh by key and token", zap.String("token", rl.token), zap.String("key", rl.key)) lockTime := time.Duration(rl.lockLeaseTime/3) * time.Second timer := time.NewTimer(lockTime) defer timer.Stop() for { select { case <-timer.C: // extend key lease time res := rl.client.Eval(luascript.RefreshLockScript, []string{rl.key, rl.rwTimeoutPrefix}, rl.lockLeaseTime, rl.token) val, err := res.Int() if err != redis.Nil && err != nil { rl.logger.Info("read lock refresh failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err)) return } if constant.RedisCode(val) == constant.RefreshLockFailure { rl.logger.Error("read lock refreash failed,can not find the read lock by key and token", zap.String("token", rl.token), zap.String("key", rl.key)) break } if constant.RedisCode(val) == constant.RefreshLockSuccess { rl.logger.Info("read lock refresh success by key and token", zap.String("token", rl.token), zap.String("key", rl.key)) } timer.Reset(lockTime) case <-rl.exit: break } } } func (rl *RedissionRWLocker) UnRLock() error { res := rl.client.Eval(luascript.UnRLockScript, []string{rl.key, rl.rwTimeoutPrefix, rl.writeWaitChanKey}, 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)) return fmt.Errorf("unlock read lock failed:%w", constant.NewRedisResult(constant.UnknownInternalError, constant.LockType, err.Error())) } if (constant.RedisCode(val) == constant.UnLockSuccess) || (constant.RedisCode(val) == constant.UnRLockSuccess) { if rl.needRefresh { rl.cancelRefreshLockTime() } rl.logger.Info("unlock read lock success", zap.String("token", rl.token), zap.String("key", rl.key)) return nil } if constant.RedisCode(val) == constant.UnRLockFailureWithWLockOccupancy { rl.logger.Info("unlock read lock failed", zap.String("token", rl.token), zap.String("key", rl.key)) return fmt.Errorf("unlock read lock failed:%w", constant.NewRedisResult(constant.UnRLockFailureWithWLockOccupancy, constant.UnLockType, "")) } return nil } func (rl *RedissionRWLocker) WLock(timeout ...time.Duration) error { if rl.exit == nil { rl.exit = make(chan struct{}) } result := rl.tryWLock().(*constant.RedisResult) if result.Code == constant.UnknownInternalError { rl.logger.Error(result.OutputResultMessage()) return fmt.Errorf("get write lock failed:%w", result) } if (result.Code == constant.LockSuccess) && rl.needRefresh { rl.once.Do(func() { // async refresh lock timeout unitl receive exit singal go rl.refreshLockTimeout() }) return nil } subMsg := make(chan struct{}, 1) defer close(subMsg) sub := rl.client.Subscribe(rl.writeWaitChanKey) defer sub.Close() go rl.subscribeLock(sub, subMsg) if len(timeout) > 0 && timeout[0] > 0 { acquireTimer := time.NewTimer(timeout[0]) for { select { case _, ok := <-subMsg: if !ok { err := errors.New("failed to read the write lock waiting for for the channel message") rl.logger.Error("failed to read the read lock waiting for for the channel message") return err } result := rl.tryWLock().(*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 } if result.Code == constant.LockSuccess { rl.logger.Info(result.OutputResultMessage()) return nil } case <-acquireTimer.C: err := errors.New("the waiting time for obtaining the write lock operation has timed out") rl.logger.Info("the waiting time for obtaining the write lock operation has timed out") return err } } } return fmt.Errorf("lock write lock failed:%w", result) } func (rl *RedissionRWLocker) tryWLock() error { lockType := constant.LockType res := rl.client.Eval(luascript.WLockScript, []string{rl.key, rl.rwTimeoutPrefix}, rl.lockLeaseTime, rl.token) val, err := res.Int() if err != redis.Nil && err != nil { return constant.NewRedisResult(constant.UnknownInternalError, lockType, err.Error()) } return constant.NewRedisResult(constant.RedisCode(val), lockType, "") } func (rl *RedissionRWLocker) UnWLock() error { res := rl.client.Eval(luascript.UnWLockScript, []string{rl.key, rl.rwTimeoutPrefix, rl.waitChanKey}, unlockMessage, rl.token) val, err := res.Int() if err != redis.Nil && err != nil { rl.logger.Info("unlock write lock failed", zap.String("token", rl.token), zap.String("key", rl.key), zap.Error(err)) return fmt.Errorf("unlock write lock failed:%w", constant.NewRedisResult(constant.UnknownInternalError, constant.UnLockType, err.Error())) } if constant.RedisCode(val) == constant.UnLockSuccess { if rl.needRefresh { rl.cancelRefreshLockTime() } rl.logger.Info("unlock write lock success", zap.String("token", rl.token), zap.String("key", rl.key)) return nil } if (constant.RedisCode(val) == constant.UnWLockFailureWithRLockOccupancy) || (constant.RedisCode(val) == constant.UnWLockFailureWithWLockOccupancy) { rl.logger.Info("unlock write lock failed", zap.String("token", rl.token), zap.String("key", rl.key)) return fmt.Errorf("unlock write lock failed:%w", constant.NewRedisResult(constant.RedisCode(val), constant.UnLockType, "")) } return nil } func GetRWLocker(client *redis.Client, ops *RedissionLockConfig) *RedissionRWLocker { r := &redissionLocker{ token: uuid.New().String(), client: client, exit: make(chan struct{}), logger: logger.GetLoggerInstance(), } if len(ops.Prefix) <= 0 { ops.Prefix = "redission-rwlock" } if len(ops.ChanPrefix) <= 0 { ops.ChanPrefix = "redission-rwlock-channel" } if ops.LockLeaseTime == 0 { r.lockLeaseTime = internalLockLeaseTime } r.key = strings.Join([]string{ops.Prefix, ops.Key}, ":") rwLocker := &RedissionRWLocker{ redissionLocker: *r, writeWaitChanKey: strings.Join([]string{r.key, "write"}, ":"), rwTimeoutPrefix: "rwlock_timeout", needRefresh: true, } return rwLocker }