所用代码地址
https://github.com/Wasabi1234/mmall

Redis分布式锁

分布式锁在很多场景中是非常有用的原语， 不同的进程必须以独占资源的方式实现资源共享就是一个典型的例子。

有很多分布式锁的库和描述怎么实现分布式锁管理器（DLM)的博客,但是每个库的实现方式都不太一样，很多库的实现方式为了简单降低了可靠性，而有的使用了稍微复杂的设计。

一种算法，叫Redlock,我们认为这种实现比普通的单实例实现更安全

安全和活性失效保障

按照思路和设计方案，算法只需具备3个特性就可以实现一个最低保障的分布式锁。

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

为什么基于故障转移的实现还不够

先分析一下当前大多数基于Redis的分布式锁现状和实现方法.

最简单的方法

就是在Redis中创建一个key，这个key有一个失效时间（TTL)，以保证锁最终会被自动释放掉（这个对应特性2）。当客户端释放资源(解锁）的时候，会删除掉这个key。

image.png

集群中各个节点都使用共享的缓存、队列，有些场景中各个节点之间可能会发生资源竞争，可能会发各个节点之间的“线程不安全问题”，
单机中，可以使用锁来解决
在分布式环境下，就要用到分布式锁

Redis分布式锁防死锁

配置锁超时时间

定义锁名

从表面上看，似乎效果还不错，但是这里有一个问题：这个架构中存在一个严重的单点失败问题。如果Redis挂了怎么办？你可能会说，可以通过增加一个slave节点解决这个问题。但这通常是行不通的。这样做，我们不能实现资源的独享,因为Redis的主从同步通常是异步的。

在这种场景（主从结构）中存在明显的竞态:

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

有时候程序就是这么巧，比如说正好一个节点挂掉的时候，多个客户端同时取到了锁。如果你可以接受这种小概率错误，那用这个基于复制的方案就完全没有问题。否则的话，我们建议你实现下面描述的解决方案。

单Redis实例实现分布式锁的正确方法

在尝试克服上述单实例设置的限制之前，让我们先讨论一下在这种简单情况下实现分布式锁的正确做法，实际上这是一种可行的方案，尽管存在竞态，结果仍然是可接受的，另外，这里讨论的单实例加锁方法也是分布式加锁算法的基础。

获取锁使用命令:

SET resource_name my_random_value NX PX 30000

这个命令仅在不存在key的时候才能被执行成功（NX选项），并且这个key有一个30秒的自动失效时间（PX属性）。这个key的值是“my_random_value”(一个随机值），这个值在所有的客户端必须是唯一的，所有同一key的获取者（竞争者）这个值都不能一样。

value的值必须是随机数主要是为了更安全的释放锁，释放锁的时候使用脚本告诉Redis:只有key存在并且存储的值和我指定的值一样才能告诉我删除成功。可以通过以下Lua脚本实现：

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

这样释放锁可以避免删除别的客户端获取成功的锁
举个例子：客户端A取得资源锁，但是紧接着被一个其他操作阻塞了，当A运行完毕后要释放锁时，原来的锁早已超时并且被Redis自动释放，
且在这期间资源锁又被客户端B再次获取到。
如果仅使用DEL命令将key删除，那么这种情况就会把客户端B的锁给删除掉。
使用Lua脚本就不会存在这种情况，因为脚本仅会删除value等于客户端A的value的key（value相当于客户端的一个签名）

这个随机字符串应该怎么设置？只要这个数在你的任务中是唯一的就行。一种简单的方法是把以毫秒为单位的unix时间和客户端ID拼接起来，理论上不是完全安全，但是在多数情况下可以满足需求.

key的失效时间，被称作“锁定有效期”。它不仅是key自动失效时间，而且还是一个客户端持有锁多长时间后可以被另外一个客户端重新获得。

截至到目前，我们已经有较好的方法获取锁和释放锁。基于Redis单实例，假设这个单实例总是可用，这种方法已经足够安全。
现在让我们扩展一下，假设Redis没有总是可用的保障。

Redlock算法

在Redis的分布式环境中，我们假设
N个Redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制
之前我们已经描述了在Redis单实例下怎么安全地获取和释放锁。我们确保将在每（N)个实例上使用此方法获取和释放锁。在这个样例中，我们假设有5个Redis master节点，这是一个比较合理的设置，所以我们需要在5台机器上面或者5台虚拟机上面运行这些实例，这样保证他们不会同时宕掉。

为了取锁，客户端应执行以下操作

• 1. 获取当前Unix时间(ms)
• 1. 依次尝试从N个实例，使用相同的key和随机值获取锁。
     在步骤2，当向Redis设置锁时,客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。
     例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试另外一个Redis实例。
• 1. 客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
• 4.如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。
• 5.如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功）

这个算法是异步的么?

基于这样一个假设：虽然多个进程之间没有时钟同步，但每个进程都以相同的时钟频率前进，时间差相对于失效时间来说几乎可以忽略不计。这种假设和我们的真实世界非常接近：每个计算机都有一个本地时钟，我们可以容忍多个计算机之间有较小的时钟漂移。

从这点来说，我们必须再次强调我们的互相排斥规则：只有在锁的有效时间（在步骤3计算的结果）范围内客户端能够做完它的工作，锁的安全性才能得到保证（锁的实际有效时间通常要比设置的短，因为计算机之间有时钟漂移的现象）。

失败时重试

当客户端无法取到锁时，应该在一个随机延迟后重试,防止多个客户端在同时抢夺同一资源的锁（这样会导致脑裂，没有人会取到锁）。同样，客户端取得大部分Redis实例锁所花费的时间越短，脑裂出现的概率就会越低（必要的重试），所以，理想情况一下，客户端应该同时（并发地）向所有Redis发送SET命令。

需要强调，当客户端从大多数Redis实例获取锁失败时，应该尽快地释放（部分）已经成功取到的锁，这样其他的客户端就不必非得等到锁过完“有效时间”才能取到（然而，如果已经存在网络分裂，客户端已经无法和Redis实例通信，此时就只能等待key的自动释放了，等于被惩罚了）。

释放锁

释放锁比较简单，向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁.

安全争议

这个算法安全么？我们可以从不同的场景讨论一下。

让我们假设客户端从大多数Redis实例取到了锁。所有的实例都包含同样的key，并且key的有效时间也一样。然而，key肯定是在不同的时间被设置上的，所以key的失效时间也不是精确的相同。我们假设第一个设置的key时间是T1(开始向第一个server发送命令前时间），最后一个设置的key时间是T2(得到最后一台server的答复后的时间），我们可以确认，第一个server的key至少会存活 MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT。所有其他的key的存活时间，都会比这个key时间晚，所以可以肯定，所有key的失效时间至少是MIN_VALIDITY。

当大部分实例的key被设置后，其他的客户端将不能再取到锁，因为至少N/2+1个实例已经存在key。所以，如果一个锁被（客户端）获取后，客户端自己也不能再次申请到锁(违反互相排斥属性）。

然而我们也想确保，当多个客户端同时抢夺一个锁时不能两个都成功。

如果客户端在获取到大多数redis实例锁，使用的时间接近或者已经大于失效时间，客户端将认为锁是失效的锁，并且将释放掉已经获取到的锁，所以我们只需要在有效时间范围内获取到大部分锁这种情况。在上面已经讨论过有争议的地方，在MIN_VALIDITY时间内，将没有客户端再次取得锁。所以只有一种情况，多个客户端会在相同时间取得N/2+1实例的锁，那就是取得锁的时间大于失效时间（TTL time)，这样取到的锁也是无效的.

/**
     * 每1分钟(每个1分钟的整数倍)
     */
    @Scheduled(cron="0 */1 * * * ?")
    public void closeOrdersTask() {
        log.info( "关闭订单定时任务启动" );

        long lockTimeout=Long.parseLong( PropertiesUtil.getProperty( "lock.timeout", "5000" ) );

        //此时有效期无限,通过setnx原子性保证了同步
        Long setnxResult=RedisSharedPoolUtil.setnx( Const.RedisLock.CLOSE_ORDER_TASK_LOCK, String.valueOf( System.currentTimeMillis() + lockTimeout ) );

        //尝试获取锁,相当于Redission中的tryLock()
        if (setnxResult != null && setnxResult.intValue() == 1) {
            //成功获得锁,设置锁有效期
            closeOrder( Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
        } else {
            //未获取到锁，继续判断，判断时间戳，看是否可以重置并获取到锁
            //该锁是否已经过期。如果没有过期，将会睡眠一会，并且从一开始进行重试操作
            String lockValueStr=RedisSharedPoolUtil.get( Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
            //尝试获取锁,相当于Redission中的tryLock()
            if (lockValueStr != null && System.currentTimeMillis() > Long.parseLong( lockValueStr )) {
                //已经过期的旧值是否仍然存储中。如果是的话，会获得锁
                String getSetResult=RedisSharedPoolUtil.getSet( Const.RedisLock.CLOSE_ORDER_TASK_LOCK, String.valueOf( System.currentTimeMillis() + lockTimeout ) );
                /**
                 *  再次用当前时间戳getset
                 *  返回给定的key的旧值，->旧值判断，是否可以获取锁
                 *  当key没有旧值时，即key不存在时，返回null ->获取锁
                 *  这里我们set了一个新的value值，获取旧的值。
                 */
                //尝试获取锁,相当于Redission中的tryLock()
                if (getSetResult == null || (getSetResult != null && StringUtils.equals( lockValueStr, getSetResult ))) {
                    //真正获取到锁       //如果返回值是1，代表设置成功，获取锁
                    closeOrder( Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
                } else {
                    log.info( "没有获取到分布式锁:{}", Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
                }
            } else {
                log.info( "没有获取到分布式锁:{}", Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
            }
        }
        log.info( "关闭订单定时任务结束" );
    }

 /**
     * 设置锁的有效期
     * 因为默认TTL无限,防止setnx持续返回0
     *
     * @param lockName 分布式锁
     */
    private void closeOrder(String lockName) {
        //有效期5秒，防止死锁
        RedisSharedPoolUtil.expire( lockName, 5 );

        log.info( "获取{},ThreadName:{}", Const.RedisLock.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName() );

        //2小时关闭订单
        int hour=Integer.parseInt( PropertiesUtil.getProperty( "close.order.task.time.hour", "2" ) );
        iOrderService.closeOrder( hour );

        //即使待删订单很少,在锁有效期内还剩余时间,关闭订单后,为确保效率,需要主动释放锁
        RedisSharedPoolUtil.del( Const.RedisLock.CLOSE_ORDER_TASK_LOCK );
        log.info( "释放{},ThreadName:{}", Const.RedisLock.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName() );
        log.info( "===============================" );
    }