用分布式锁解决并发问题

在系统中，当存在多个进程和线程可以改变某个共享数据时，就容易出现并发问题导致共享数据的不一致性。即多个进程同时获取到了对数据的操作权限并对数据进行了更新，很典型的场景就是在线销售系统在售卖热销商品时遇到多个并发请求在同一时间提交订单的情况则极有可能造成商品超卖的现象。只要访问流量不错的系统都有可能遭遇并发请求造成数据库中数据重复写入的情况。

针对程序块被多个进程并发执行问题的解决方案是确保同一个时刻同一个程序块只能有一个进程可执行，其他进程等待当前进程执行完成才能获取程序块的执行权对数据进行更新，以此类推将并发执行变为串行顺序执行。为了让获取执行权的进程不被其他干扰，就需要设置一个所有进程都能读取到的标记，当标记不存在时可以设置该标记，其余后续进程发现已经有标记了则等待拥有标记的进程结束执行程序块取消标记后再去尝试设置标记。这个标记可以理解为锁，设置标记的过程就是我们通常说的加锁。

用redis 的 setnx、expire 方法做分布式锁

• setnx()

setnx 的含义就是 SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。

• expire()

expire 设置过期时间，要注意的是 setnx 命令不能设置 key 的超时时间，只能通过 expire() 来对 key 设置。

• 具体步骤

1、setnx(lockKey, 1) 如果返回 0，则说明占位失败；如果返回 1，则说明占位成功

2、expire() 命令对 lockKey 设置超时时间，为的是避免死锁问题。

3、执行完业务代码后，可以通过 delete 命令删除 key。

这个方案其实是可以解决日常工作中的需求的，但从技术方案的探讨上来说，可能还有一些可以完善的地方。比如，如果在第一步 setnx 执行成功后，在 expire() 命令执行成功前，发生了宕机的现象，那么就依然会出现死锁的问题，所以如果要对其进行完善的话，可以使用 redis 的 setnx()、get() 和 getset() 方法来实现分布式锁。

用 redis 的 setnx()、get()、getset()方法做分布式锁

这个方案的背景主要是在 setnx() 和 expire() 的方案上针对可能存在的死锁问题，做了一些优化。

• getset()

这个命令主要有两个参数 getset(key，newValue)。该方法是原子的，对 key 设置 newValue 这个值，并且返回 key 原来的旧值。假设 key 原来是不存在的，那么多次执行这个命令，会出现下边的效果：

1. getset(key, "value1") 返回 null 此时 key 的值会被设置为 value1
2. getset(key, "value2") 返回 value1 此时 key 的值会被设置为 value2
3. 依次类推！

• 使用步骤

1. setnx(lockKey, 当前时间+过期超时时间)，如果返回 1，则获取锁成功；如果返回 0 则没有获取到锁，转到步骤 2。
2. get(lockKey) 获取值，值是当前lockKey的过期时间用oldExpireTime代表 ，并将这个 oldExpireTime与当前的系统时间进行比较，如果早于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向 步骤3，否则等待指定时间后返回步骤2重新开始判定。
3. 计算 newExpireTime = 当前时间+过期超时时间，然后 getset(lockKey, newExpireTime) 会返回当前 lockKey 之前设置的旧值currentExpireTime。
4. 判断 currentExpireTime 与 oldExpireTime 是否相等，如果相等，说明当前进程getset 设置锁成功，获取到了锁。如果不相等，说明这个锁已经被别的进程获取走了，那么当前请求可以根据具体需求逻辑直接返回失败，或者返回步骤2继续重试。
5. 在获取到锁之后，当前进程可以开始自己的业务处理，当处理完毕后，比较当前理时间和对锁设置的超时时间，如果小于锁设置的超时时间，则直接执行 delete 释放锁；如果大于锁设置的超时时间，锁可能已由其他进程获得，这时执行 delete释放锁的操作会导致把其他进程已获得的锁释放掉。

下面是用PHP代码实现的Redis分布式锁，关于Redis部分使用的是伪代码，请根据自己的情况用Redis连接对象替代其中的伪代码。

/**
 * 获取Redis分布式锁
 *
 * @param $lockKey
 * @return bool
 */
function getRedisDistributedLock(string $lockKey) : bool
{
    $lockTimeout = 2000;// 锁的超时时间2000毫秒
    $now = intval(microtime(true) * 1000);
    $lockExpireTime = $now + $lockTimeout;
    $lockResult = Redis::setnx($lockKey, $lockExpireTime);

    if ($lockResult) {
        // 当前进程设置锁成功
        return true;
    } else {
        $oldLockExpireTime = Redis::get($lockKey);
        if ($now > $oldLockExpireTime && $oldLockExpireTime == Redis::getset($lockKey, $lockExpireTime)) {
            return true;
        }
    }

    return false;
}

/**
 * 串行执行程序
 *
 * @param string $lockKey Key for lock
 * @param Closure $closure 获得锁后进程要执行的闭包
 * @return mixed
 */
function serialProcessing(string $lockKey, Closure $closure)
{
    if (getRedisDistributedLock($lockKey)) {
        $result = $closure();
        $now = intval(microtime(true) * 1000);
        if ($now < Redis::get($lockKey)) {
            Redis::del($lockKey);   
        }
    } else {
        // 延迟200毫秒再执行
        usleep(200 * 1000);
        return serialProcessing($lockKey, $closure);
    }

    return $result;
}

上面serialProcessing方法里当前进程设置锁成功，获取了代码块的执行权后就会执行闭包参数$closure里的代码块，通过传递闭包给方法，让我们可以在项目任何需要确保程序串行执行的地方使用serialProcessing方法给程序加分布式锁解决并发请求的问题。

上面代码实现用面向过程的方式是为了能简单明了的描述怎么设置分布式锁，读者可以针对自己的情况执行设计实现代码。针对于大型系统使用集群Redis的情况，设置分布式锁的步骤更复杂，有兴趣的可以查看Redlock 算法和redissonredis分布式锁组件。