Redis分布式锁
Redis 分布式锁是一种使用 Redis 数据库实现分布式锁的方式,可以保证在分布式环境中同一时间只有一个实例可以访问共享资源。
实现机制
以下是实现其加锁步骤:
获取锁
在 Redis 中,一个相同的key
代表一把锁。是否拥有这把锁,需要判断key
和value
是否是自己设置的,同时还要判断锁是否已经过期。
- 首先通过get命令去获取锁,如果获取不到说明还没有加锁
- 如果还没有加锁我们就可以去通过set命令去加锁,并且需要设置一个expire过期时间防止成为一个长生不老锁,那如果业务还没有执行完锁就释放了怎么办呢?这个后面会提到续锁
- 如果获取到了key说明已经被其他实例抢到了锁,加锁失败
- 加锁失败还需要根据一些操作例如超时时间内去重试加锁,直到加锁成功或者超时
这些操作都需要原子性操作,需要用lua脚本进行封装
lock.lua
val = redis.call('get', KEYS[1])
if val == false then
return redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2])
elseif val == ARGV[1] then
redis.call('expire', KEYS[1], ARGV[2])
return 'OK'
else
return ''
end
释放锁
释放锁的时候就是把key删除,不过删除的时候需要判断是不是自己加的锁
unlock.lua
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
Go 实现分布式锁
结构体字段配置
// redis客户端连接
type Client struct {
client redis.Cmdable
varFunc func() string
g singleflight.Group
}
// 锁的结构体
type Lock struct {
client redis.Cmdable
key string
value string
expiration time.Duration
unlock chan struct{}
unlockOne sync.Once
}
// NewClient creates a *Client
func NewClient(client redis.Cmdable) *Client {
return &Client{
client: client,
varFunc: func() string {
return uuid.New().String()
},
}
}
// 重试策略
type RetryStrategy interface {
// Next determines the time interval for Lock
// and whether Lock to retry
Next() (time.Duration, bool)
}
// 周期性重试
type FixedIntervalRetry struct {
Interval time.Duration
Max int
cnt int
}
lua 脚本,使用go的embed映射到luaLock string
var (
ErrFailedToPreemptLock = errors.New("redis-lock: failed to lock")
ErrLockNotHold = errors.New("redis-lock: lock not hold")
ErrLockTimeout = errors.New("redis-lock: lock timeout")
//go:embed lua/unlock.lua
luaUnlock string
//go:embed lua/refresh.lua
luaRefresh string
//go:embed lua/lock.lua
luaLock string
)
加锁Lock
加锁时有两种方案,一种是比较简单的( TryLock )尝试加锁,只需要传个过期时间,另一种是比较完善的( Lock )加锁,会有超时策略等
func newLock(client redis.Cmdable, key string, value string, expiration time.Duration) *Lock {
return &Lock{
client: client,
key: key,
value: value,
expiration: expiration,
unlock: make(chan struct{}, 1),
}
}
// TryLock tries to acquire a lock
func (c *Client) TryLock(ctx context.Context,
key string,
expiration time.Duration) (*Lock, error) {
val := c.varFunc()
ok, err := c.client.SetNX(ctx, key, val, expiration).Result()
if err != nil {
return nil, err
}
if !ok {
return nil, ErrFailedToPreemptLock
}
return newLock(c.client, key, val, expiration), nil
}
// Lock tries to acquire a lock with timeout and retry strategy
func (c *Client) Lock(ctx context.Context,
key string,
expiration time.Duration,
timeout time.Duration, retry RetryStrategy) (*Lock, error) {
var timer *time.Timer
val := c.varFunc()
for {
lCtx, cancel := context.WithTimeout(ctx, timeout)
res, err := c.client.Eval(lCtx, luaLock, []string{key}, val, expiration.Seconds()).Result()
cancel()
if err != nil && !errors.Is(err, context.DeadlineExceeded) {
return nil, err
}
if res == "OK" {
return newLock(c.client, key, val, expiration), nil
}
interval, ok := retry.Next()
if !ok {
return nil, ErrLockTimeout
}
if timer == nil {
timer = time.NewTimer(interval)
} else {
timer.Reset(interval)
}
select {
case <-timer.C:
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
解锁unLock
// Unlock releases the lock
func (l *Lock) Unlock(ctx context.Context) error {
res, err := l.client.Eval(ctx, luaUnlock, []string{l.key}, l.value).Int64()
defer func() {
l.unlockOne.Do(func() {
l.unlock <- struct{}{}
close(l.unlock)
})
}()
if errors.Is(err, redis.Nil) {
return ErrLockNotHold
}
if err != nil {
return err
}
if res != 1 {
return ErrLockNotHold
}
return nil
}
小结
- 使用分布式锁本身会有各种各样的问题,需要自己去处理异常情况例如超时等
- 对锁的操作一定要判断是不是自己加的那把锁,否则会误删会导致业务错误
- 对锁的续约部分我们下一篇再讲
本文go的代码是完整的,可以直接copy使用,有兴趣的小伙伴可以去使用一下