Redis 分布式锁

代码地址：https://github.com/xuqil/redis-lock

Redis 分布式锁是一种使用 Redis 数据库实现分布式锁的方式，可以保证在分布式环境中同一时间只有一个实例可以访问共享资源。

实现机制

Redis 分布式锁中需要用到的命令：

• SET key value [EX seconds | PX milliseconds]：设置带过期时间的key-value。
• EXPIRE key seconds：给指定的key设置过期时间。
• GET key：获取给定的key的value。
• DEL key：删除给定的key。
• SETNX key value：如果key不存在，则设置key-value，反正设置失败。这里用SET和GET一起使用代替。

获取锁

在 Redis 中，一个相同的key代表一把锁。是否拥有这把锁，需要判断key和value是否是自己设置的，同时还要判断锁是否已经过期。

以下是某个实例加锁的步骤：

1. 通过GET命令获取key，如果获取不到key，说明还没有加锁；
2. 如果没有加锁，则使用SET命令设置key，同时设置锁的过期时间，加锁成功。返回；
3. 如果获取到了key，并且value是自己设置的，证明该实例已经加锁成功，此时需要使用EXPIRE命令为锁添加过期时间，因为这次可能是重试，前一次已经加锁成功。返回；
4. 如果获取到了key，但是value不属于自己设置的，证明已经被其他实例抢到了锁，加锁失败。
5. 加锁失败，则继续进行 1~4 步骤，直至超时或者加锁成功。

以上 1~4 步骤需要原子性操作，可以通过 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

释放锁

1. 主动释放：释放锁其实就是删除key，使用DEL命令进行删除。删除key前，需要判断key对应的value是否为自己设置的value，如果不是，证明锁已经被其他实例获取。判断和删除都也需要是原子操作。

   if redis.call('get', KEYS[1]) == ARGV[1] then
       return redis.call('del', KEYS[1])
   else
       return 0
   end
   

2. 过期释放：由于锁（即key）设置了过期，如果锁没有被续期（增加过期时间），就会被 Redis 删除。

需要注意的是，使用 Redis 实现分布式锁需要考虑一些问题，例如 Redis 实例的可用性、网络延迟、锁的持有者异常退出等，需要进行合理的设计和实现。另外，为了保证锁的正确性和可靠性，可以采用一些常用的技术手段，例如设置合适的超时时间、使用 RedLock 算法、采用 Lua 脚本等。

Go 实现 Redis 分布式锁

加锁

// 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)
       // 这里通过上面的 lua 脚本尝试获取锁，luaLock
      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()
      }
   }
}

以上代码通过 lua 脚本实现了可重试的机制，如果没有获取到锁，且在超时前就一直尝试获取锁，获取锁成功后就返回一下*Lock结构体，Lock实现了释放锁的方法Unlock：

type Lock struct {
	client     redis.Cmdable
	key        string
	value      string
	expiration time.Duration
	unlock     chan struct{}
	unlockOne  sync.Once
}

解锁

// 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 err == redis.Nil {
		return ErrLockNotHold
	}
	if err != nil {
		return err
	}
	if res != 1 {
		return ErrLockNotHold
	}
	return nil
}

跟加锁一样，解锁也是通过 lua 脚本来实现。释放锁的时候需要注意：锁是不是自己的，可以通过key和value来判断。

续约

锁的过期时间应该设置多长？

• 设置短了，那么业务还没完成，锁就过期了。
• 设置长了，万一实例崩溃了，那么其它实例也长时间拿不到锁。

更严重的是，不管你设置多长，极端情况下，都会出现业务执行时间超过过期时间。

我们可以考虑在锁还没有过期的时候，再一次延长过期时间，那么：

• 过期时间不必设置得很长，自动续约会帮我们设置好。
• 如果实例崩溃了，则没有人再续约，过一段时间之后自然就会过期，其它实例就能拿到锁了。

续约其实就是对 Redis 的key延长过期时间，需要注意的时，续期也要判断锁是不是自己的，因为锁可能已经过期被其他实例获取了。

// Refresh refreshes the lock by expiration
func (l *Lock) Refresh(ctx context.Context) error {
   res, err := l.client.Eval(ctx, luaRefresh, []string{l.key}, l.value, l.expiration.Seconds()).Int64()
   if err != nil {
      return err
   }
   if res != 1 {
      return ErrLockNotHold
   }
   return nil
}

singleflight 优化

在非常高并发并且热点集中的情况下，可以考虑结合 singleflight 来进行优化。也就是说，本地所有的 goroutine 自己先竞争一把，胜利者再去抢全局的分布式锁。

func (c *Client) SingleflightLock(ctx context.Context,
   key string,
   expiration time.Duration,
   timeout time.Duration, retry RetryStrategy) (*Lock, error) {
   for {
      var flag bool
      resCh := c.g.DoChan(key, func() (interface{}, error) {
         flag = true
         return c.Lock(ctx, key, expiration, timeout, retry)
      })
      select {
      case res := <-resCh:
         if flag {
            c.g.Forget(key)
            if res.Err != nil {
               return nil, res.Err
            }
            return res.Val.(*Lock), nil
         }
      case <-ctx.Done():
         return nil, ctx.Err()
      }
   }
}

Redlock

前面讨论的都是单点的 Redis，在集群部署的时候，需要额外考虑一个问题：主从切换。

一切顺利的情况：

主从切换的异常情况：

Redlock 的思路：不再部署单一主从集群，而是多个主节点（没有从节点）。

比如说我们部署五个主节点，那么加锁过程是类似的，只是要在五个主节点上都加上锁，如果多数（这里是三个）都成功了，那么就认为加锁成功。

优化

Redis 分布式锁是实现分布式锁的一种常用方式，以下是一些可以优化 Redis 分布式锁的方法：

1. 使用 RedLock 算法：在 Redis 分布式锁中，为了防止发生死锁，可以使用 RedLock 算法。这种算法是将锁分配到多个 Redis 实例上，通过协同工作来实现分布式锁的目的。当某个 Redis 实例无法正常工作时，其他实例可以继续提供服务，从而避免出现死锁的情况。
2. 降低 Redis 的网络延迟：在使用 Redis 分布式锁时，网络延迟可能会导致性能问题。可以通过降低 Redis 的网络延迟来提高性能，例如使用本地的 Redis 实例，或者使用高速网络。
3. 减少 Redis 的操作：在使用 Redis 分布式锁时，应该尽量减少 Redis 的操作次数，以提高性能。例如，可以将锁的持有者信息存储在本地内存中，而不是每次都从 Redis 中获取。
4. 使用超时时间：在获取 Redis 分布式锁时，应该设置一个超时时间，以避免出现死锁的情况。当一个客户端获取锁后，在规定的时间内未能释放锁，其他客户端可以将其锁定的键值对删除，从而让其他客户端获取锁。
5. 使用 Lua 脚本：Lua 脚本是 Redis 内置的一种脚本语言，可以用来实现一些复杂的操作，例如分布式锁。通过使用 Lua 脚本，可以将多个 Redis 操作封装成一个原子操作，从而提高性能和安全性。
6. 使用 Sentinel 高可用方案：Sentinel 是 Redis 的高可用方案之一，它可以监控 Redis 实例的健康状态，并在发生故障时自动切换到备用实例。通过使用 Sentinel，可以提高 Redis 分布式锁的可用性和稳定性。

小结

• 使用分布式锁，你不能指望框架提供万无一失的方案，自己还是要处理各种异常情况（超时）。
• 自己写分布式锁，要考虑过期时间，以及要不要续约。
• 不管要对锁做什么操作，首先要确认这把锁是我们自己的锁。
• 多数时候，与其选择复杂方案，不如直接让业务失败，可能成本还要低一点：有时候直接赔钱，比你部署一大堆节点，招一大堆开发，搞好几个机房还要便宜，而且便宜很多。
• 选择恰好的方案，而不是完美的方案。

参考

• Distributed Locks with Redis (opens new window)