Redis实现分布式锁

时间 2020-07-06

标签 redis 实现分布式栏目 Redis 繁體版

原文原文链接

分布式锁的实现方式有不少，本篇文章讲述一下使用Redis实现分布式锁。网上有不少使用Redis实现分布式锁的代码，可是这些代码或多或少都有问题。这篇文章会写一个实现，同时标明一些注意点。php

场景

为了便于阐述，这里假设一个游戏场景，用户A有开山斧一把，价值500元宝，用户B有800元宝，想买A的开山斧，这些数据都存在Redis中。须要编写代码成功的实现该笔交易。html

问题

Redis实现分布式锁，须要考虑以下问题：程序员

持有锁的进程由于操做时间过长而致使锁被自动释放，但进程自己并不知晓这一点，甚至还可能会错误地释放掉了其余进程持有的锁。
一个持有锁并打算执行长时间操做的进程已经崩溃，但其余想要获取锁的进程不知道哪一个进程持有着锁，也没法检测出持有锁的进程已经崩溃，只能白白地浪费时间等待锁被释放。
在一个进程持有的锁过时以后，其余多个进程同时尝试去获取锁，而且都得到了锁。

三个特性

实现一个最低保障的分布式锁，须要具有三个特性redis

安全属性（Safety property）: 独享（相互排斥）。在任意一个时刻，只有一个客户端持有锁。
活性A(Liveness property A): 无死锁。即使持有锁的客户端崩溃（crashed)或者网络被分裂（gets partitioned)，锁仍然能够被获取。
活性B(Liveness property B): 容错。只要大部分Redis节点都活着，客户端就能够获取和释放锁.

命令

使用Redis实现分布式锁，通常使用SETNX或者SET命令，SETNX不能同时设置过时时间，若是使用的版本大于等于2.6.12，可使用SET命令，可使用这个命令原子性的实现SETNX和EXPIRE的功能，下面是两个命令的简介算法

SETNX

命令格式：SETNX key value缓存

时间复杂度：O(1)安全

说明：将key设置值为value，若是key不存在，这种状况下等同SET命令。当key存在时，什么也不作。SETNX是”SET if Not eXists”的简写。网络

返回值框架

1 若是key被设置了
0 若是key没有被设置

SET

命令格式：SET key value [EX seconds] [PX milliseconds] [NX|XX]编辑器

时间复杂度：O(1)

说明：将键key设定为指定的“字符串”值。若是 key 已经保存了一个值，那么这个操做会直接覆盖原来的值，而且忽略原始类型。当set命令执行成功以后，以前设置的过时时间都将失效。

选项

从2.6.12版本开始，redis为SET命令增长了一系列选项:

EX seconds – 设置键key的过时时间，单位时秒
PX milliseconds – 设置键key的过时时间，单位是毫秒
NX – 只有键key不存在的时候才会设置key的值
XX – 只有键key存在的时候才会设置key的值

实现

此处使用SETNX实现，毕竟有的公司Redis版本可能较低，使用SETNX能够实现，SET更加没有问题。

代码以下：

<?php

function uuid($prefix = '') {
    $chars = md5(uniqid(mt_rand(), true));
    $uuid  = substr($chars, 0, 8) . '-';
    $uuid .= substr($chars, 8, 4) . '-';
    $uuid .= substr($chars, 12, 4) . '-';
    $uuid .= substr($chars, 16, 4) . '-';
    $uuid .= substr($chars, 20, 12);
    $ret = $prefix . $uuid;
    return strtoupper($ret);
}

function acquireLock($redis,$lockName, $acquireTime = 10, $lockTime = 10) {
    $lockKey    = 'lock:' + $lockName;
    $identifier = uuid('identify');
    $end        = time() + $acquireTime;
    while (time() < $end) {
        if ($redis->setnx($lockKey, $identifier)) {
            $redis->expire($lockKey, $lockTime);
            return $identifier;
        } elseif ($redis->ttl($lockKey) == -1) {
            $redis->expire($lockKey, $lockTime);
        }
        usleep(1000);
    }
    return false;
}


function process(){
    $redis      = new Redis();
    $lockName = 'market';
    //1.获取锁
    $locked = acquireLock($redis,$lockName);
    if($locked === false){
        return false;
    }
    //2.进行交易
    //判断A和B是否知足交易条件
    //使用管道，对A和B进行操做

    //3.释放锁
    $releaseRes = releaseLock($redis,$lockName,$locked);
    if($releaseRes === false){
        return false;
    }
}

function releaseLock($redis,$lockName,$identifier){
    $lockKey    = 'lock:' + $lockName;
    $redis->watch($lockKey);
    if($redis->get($lockKey) === $identifier){
        $redis->multi();
        $redis->del($lockKey);
        $redis->exec();
        return true;
    }
    $redis->unwatch();
    return false;
}
复制代码

说明：

获取锁：
- 建立惟一的$identifier，这个值用于删除key的时候，判断是否为当前客户端获取的锁，以避免删除其它客户端的锁
- while循环用于在一段时间内不停的获取锁
- 若是可以获取锁便获取，同时设置超时时间，防止线程运行时崩溃，锁永远没法释放
- 若是没能成功获取到锁，检查当前锁的过时时间，若是未设置过时时间，进行设置，防止其余线程得到锁后当即崩溃，没有设置过时时间
处理业务
- 须要先判断交易双方是否都知足条件，由于锁定的是整个市场，因此一旦得到锁，交易双方的状态都不会再进行改变
- 使用管道能保证整个交易能像事务同样被处理，并且性能会比用redis的事务更好
释放锁
- 使用watch监控锁，一旦key被变动，删除key的事务不会被执行
- 须要判断key的值是否和本线程记录的$identifier同样，只有一致才能进行删除
- 使用事务来作删除key的操做，使用事务的缘由是防止中途该锁被别的线程获取
- 若是失败，记得unwatch
其余问题
- 可重入问题：可重入指的是，线程能够再次获取到锁。实现方法比较简单，只须要在acquireLock的时候，传入identifier，判断当前锁的identifier和传入的是否一致，若是一致则能够进行操做
- 线程未执行完毕，锁的超时时间已过，其余线程获取到锁：解决该问题的一个方案是，当获取到锁后，在超时时间通过一半的时候检查锁是否存在或者被修改，若是没有变化且线程正常运行，则延长超时时间

思考

若是基于Redis单实例，假设这个单实例老是可用，这种方法已经足够安全。

但有两种特殊状况你们须要关注一下：

主从结构中存在明显的竞态:

客户端A从master获取到锁
在master将锁同步到slave以前，master宕掉了。
slave节点被晋级为master节点
客户端B取得了同一个资源被客户端A已经获取到的另一个锁。安全失效！

在Redis的分布式环境中，有N个Redis master

这种状况可使用Redlock算法

总结

本文讲述了怎样用Redis实现分布式锁，并写了具体实现和相关的分析。要用Redis实现分布式锁，有不少细节须要思考，你们能够根据本身的业务形态设计符合本身要求的锁，在复杂度和安全性上作好折中。

资料

最后

你们若是喜欢个人文章，能够关注个人公众号（程序员麻辣烫）

往期文章回顾：