PHP使用Redis实现分布式锁

背景

在分布式系统中，分布式锁是确保资源互斥访问的重要机制。我们在系统中修改已有数据时，需要先读取，然后进行修改保存，此时很容易遇到并发问题。由于修改和保存不是原子操作，在并发场景下，部分对数据的操作可能会丢失。在单服务器系统我们常用本地锁来避免并发带来的问题，然而，当服务采用集群方式部署时，本地锁无法在多个服务器之间生效，这时候保证数据的一致性就需要分布式锁来实现。

实现方式

通过 Redis 的 SET 和 DEL 命令实现锁的设置和释放，并使用 Lua 脚本确保操作的原子性。

• 加锁命令：SETNX key value，当键不存在时，对键进行设置操作并返回成功，否则返回失败。KEY 是锁的唯一标识，一般按业务来决定命名。

• 解锁命令：DEL key，通过删除键值对释放锁，以便其他线程可以通过 SETNX 命令来获取锁。

• 锁超时：EXPIRE key timeout, 设置 key 的超时时间，以保证即使锁没有被显式释放，锁也可以在一定时间后自动释放，避免资源被永远锁住。

以下是详细的代码分析：

获取锁

getLock 方法尝试获取锁。它使用 Redis 的 SET 命令，并通过 Lua 脚本确保操作的原子性。关键点如下：

• 锁超时：设置 key 的超时时间，以保证即使锁没有被显式释放，锁也可以在一定时间后自动释放，避免资源被永远锁住。
• 唯一值生成：使用 md5(uniqid('', true)) 生成唯一的锁值，在后面释放锁时需要用到。
• 原子性：SET 命令结合 NX 和 EX 选项确保锁的唯一性和过期时间。Lua 脚本避免了多命令操作的并发问题。
• 重试机制：如果锁获取失败，方法会自旋重试，直到达到最大重试次数。每次重试之间，线程会等待 200 毫秒。

/**
  * 获取锁，支持自旋
  *
  * @param  string  $lockKey
  * @param  int     $ttl           过期时间
  * @param  int     $max_attempts  最大重试次数
  *
  * @return string|null
*/
public function getLock(string $lockKey, int $ttl = 20, int $max_attempts = 10): ?string
{
  $key = self::LOCK_PREFIX.$lockKey;
  $uniqueValue = md5(uniqid('', true)); // 生成唯一的锁值
  $attempts = 0;
  while ($attempts < $max_attempts) {
    $luaScript = <<<SCRIPT
            if redis.call('SET', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then
                return true
            else
                return false
            end
SCRIPT;

    $result = Redis::eval($luaScript, 1, $key, $uniqueValue, $ttl);
    if ($result) {
      return $uniqueValue; // 成功获取锁
    }
    usleep(200 * 1000); // 等待一段时间（微秒）
    $attempts++;
  }

  return null; // 达到最大重试次数后仍未获取到锁
}

释放锁

releaseLock 方法用于释放锁。它通过 Lua 脚本实现，确保只有持有锁的客户端才能释放锁。关键点如下：

• 锁误解除：如果线程 A 成功获取到了锁，并且设置了过期时间 30 秒，但线程 A 执行时间超过了 30 秒，锁过期自动释放，此时线程 B 获取到了锁；随后 A 执行完成，线程 A 使用 DEL 命令来释放锁，但此时线程 B 加的锁还没有执行完成，线程 A 实际释放的线程 B 加的锁。所以我们要获取当前锁的值并检查是否与传入的 lockValue 匹配。如果匹配，则执行 DEL 命令释放锁。这样可以防止误删其他客户端持有的锁。

public function releaseLock(string $lockKey, string $lockValue): bool
{
  $key = self::LOCK_PREFIX.$lockKey;

  $luaScript = <<<SCRIPT
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        else
            return 0
        end
SCRIPT;

  $result = Redis::eval($luaScript, 1, $key, $lockValue);

  return $result === 1; // 返回布尔值，表示是否成功释放锁
}

使用

获取了锁之后一定要释放锁，所以用try finally的错误捕获方法保证不管在获取锁之后是否发生错误，最后都会释放锁，这是安全使用锁的一种姿势。

$lockKey = 'xxx';//业务key
$lockValue = RedisLockService::getInstance()->getLock($lockKey);
if(!$lockValue) {
    //没有拿到key
    return;
}
try {
    // 拿到锁后处理业务
} finally {
    // 释放锁
    RedisLockService::getInstance()->releaseLock($lockKey, $lockValue);
}

Enjoy it