用分布式锁解决并发问题

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

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

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

• setnx()

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

• expire()

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

• 具体步骤

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

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

三、执行完业务代码后，能够经过 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分布式锁组件。