Redis分布式锁之红锁（RedLock）

前言

在许多不同的进程必须以互斥的方式操作共享资源的环境中，分布式锁是非常有用的。

我们在使用Redis做分布式锁时，使用的一般都是比较简单的方法。

本文提供一个更规范的算法来实现Redis分布式锁,名为Redlock的算法。

正文

分布式锁特点

我们知道，分布式锁应保证以下三点：

• 安全性：互斥性。在任何给定时刻，只有一个客户端可以持有一个锁。
• 可靠性 A：无死锁。最终，总是有可能获得一个锁，即使锁定资源的客户机崩溃或被分区。
• 可靠性 B：容错性。只要大部分Redis节点都是正常的，客户端就可以获取和释放锁。

问题

在Redis服务中，一般由Redis集群提供服务，设置为主从模式（Master-Slave），主从Redis之间的信息拷贝是异步完成的。

我们试想，如果当我们请求分布式锁的时候成功了，但是此时 Slave 还没有复制我们的“锁”，如果此时 Master 由于某些原因宕机，Slave服务器变为Master，我们应用继续请求锁的时候，就会成功创建。这就出现了同一个锁获取了不止一次。

这样肯定会有一些问题。

单机实例获取锁

我们先来看下如何在Redis单例模式下实现分布式锁。

获得锁的主要命令如下：

SET resource_name my_random_value NX PX 30000

该命令只在key不存在(NX选项)时设置该key，其过期时间为30000毫秒(PX选项)。该键值被设置为“myrandomvalue”。这个值必须在所有客户端和所有锁请求中是唯一的。使用随机值的目的是为了安全的释放锁。

释放锁的主要命令如下，我们可以用Lua脚本来实现：

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

为了避免删除另一个客户端创建的锁，这一点很重要。例如，一个客户端可能获得了锁，但在某些操作中阻塞的时间超过了锁的有效时间(key将过期的时间)，然后删除其他客户端已经获得的锁。所以仅仅使用DEL是不安全的，因为一个客户端可能会删除另一个客户端的锁。在上面的脚本中，每个锁都是用随机字符串“签名”的，所以只有当锁仍然是由试图删除它的客户端设置的锁时，锁才会被删除。

随机字符串的选取我们可以使用一些生成不重复字符串的加密算法，如 MD5，或者更简便的可以使用 服务器地址 + Unix 时间戳来表示。

可以看到在单机模式下，这种方案是安全可用的。

RedLock算法

在分布式版本的算法中，我们假设我们有N个Redis master。这些节点是完全独立的，所以我们不用复制等其他处理。

我们已经描述了如何在单个实例中安全地获取和释放锁。我们想当然地认为，该算法将使用该方法在单个实例中获取和释放锁。

在我们的例子中，我们设置N=5，所以我们需要在不同的计算机或虚拟机上运行5个Redis master，以确保它们以一种基本独立的方式失败。

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

1. 它以毫秒为单位获取当前时间。
2. 它尝试在所有N个实例中依次获取锁，在所有实例中使用相同的 key 和 random_value 。在步骤2中，当在每个实例中设置锁时，客户端使用一个比锁自动释放总时间小的超时来获取锁。例如，如果自动释放时间是10秒，则超时时间可以在 5~50 毫秒范围内。如果一个实例不可用，我们应该尽快尝试与下一个实例进行交互。
3. 客户端通过从当前时间中减去在步骤1中获得的时间戳来计算为了获得锁花费了多少时间。当且仅当客户端能够在大多数实例(至少3个)中获得锁，并且获取锁的总时间小于锁有效时间时，则认为该锁已被获取。
4. 如果获得了锁，则将其有效时间视为初始有效时间减去经过的时间，如步骤3中计算的那样。
5. 如果客户端由于某些原因无法获得锁（要么无法锁定N/2+1个实例（半数以上），要么有效时间为负数），它将尝试解锁所有实例（甚至是它认为自己无法锁定的实例）。

对于释放锁，很简单，只要释放所有实例中的锁，不管客户端是否认为自己能够成功锁定给定实例。

这也是RedLock的基本原理。

算法的安全性

我们假设客户端能在大多数Redis实例中获取锁，所有实例都将包含一个存在时间相同的key。但是，key是在不同的时间进行设置的，因此key的具体过期时间也不同。

但我们假设设置第一个Redis实例时时间为T1，最后一个Redis实例时时间为T2，则我们可以认为锁的有效时间 MIN_VALIDITY = TTL - (T2-T1)，即第一个实例剩余过期时间，其他实例都在后面依次过期。

可以看到在大多数key被设置的时间内，另一个客户端将无法获得锁，因为如果半数以上实例的key已经存在，那么N/2+1个set NX操作将无法成功。因此，如果获得了一个锁，就不可能同时重新获得它。

然而，我们也希望确保多个客户端同时尝试获取锁不能同时成功。

如果客户端锁定大多数实例所用的时间接近或大于锁的最大有效时间（TTL），它会认为锁无效，并将解锁实例，因此，我们只需要考虑这样一种情况：客户端能够在一段时间内锁定大多数实例，而这段时间小于有效时间。

在这种情况下，对于上面已经表示的参数 MIN_VALIDITY，没有客户端能够重新获得锁。因此，只有当锁定大多数实例的时间大于TTL时，多个客户端才能同时锁定N/2+1个实例，导致锁定无效。

关于该算法安全性及其他的一些讨论大家可以参考Redis官网的这篇文章 Distributed locks with Redis。

相关代码实现

Java里Redission就实现了RedLock算法，我们来看下。

首先 Maven 里要引入相关 Jar 包。

<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.9.0</version>
</dependency> 

代码实现如下：

public class RedisLock {
    /**
     * 获取锁最大等待时间
     */
    private static final long WAIT_TIMEOUT = 5L;
    /**
     * 锁持有时间，获取锁后超过这个时间锁会释放，该时间应大于业务逻辑处理时间
     */
    private static final long LEASE_TIME = 10L;

    private static final String LOCK_KEY = "REDISSON_LOCK_TEST";

    public static void main(String[] args) {
        Config config1 = new Config();
        config1.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456").setDatabase(0);
        RedissonClient redissonClient1 = Redisson.create(config1);

        Config config2 = new Config();
        config2.useSingleServer().setAddress("redis://127.0.0.1:6380").setPassword("123456").setDatabase(0);
        RedissonClient redissonClient2 = Redisson.create(config2);

        Config config3 = new Config();
        config3.useSingleServer().setAddress("redis://127.0.0.1:6381").setPassword("123456").setDatabase(0);
        RedissonClient redissonClient3 = Redisson.create(config3);
        //拿到RLock对象
        RLock lock1 = redissonClient1.getLock(LOCK_KEY);
        RLock lock2 = redissonClient2.getLock(LOCK_KEY);
        RLock lock3 = redissonClient3.getLock(LOCK_KEY);
        //构建redlock
        RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);

        try {
            boolean res = redLock.tryLock(WAIT_TIMEOUT, LEASE_TIME, TimeUnit.SECONDS);
            if (res) {
                //do something
            }
        } catch (Exception e) {
            throw new RuntimeException("加锁异常！！");
        }finally{
            redLock.unlock();
        }
    }

}

源码分析

我们看下redLock.tryLock方法的实现，相关源码如下：

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
//        try {
//            return tryLockAsync(waitTime, leaseTime, unit).get();
//        } catch (ExecutionException e) {
//            throw new IllegalStateException(e);
//        }
    long newLeaseTime = -1;
    if (leaseTime != -1) {
        newLeaseTime = unit.toMillis(waitTime)*2;
    }
    
    long time = System.currentTimeMillis();
    long remainTime = -1;
    if (waitTime != -1) {
        remainTime = unit.toMillis(waitTime);
    }
    long lockWaitTime = calcLockWaitTime(remainTime);
    // 允许加锁失败节点个数限制，半数以上（N-(N/2+1)）
    int failedLocksLimit = failedLocksLimit();
    // 遍历所有节点通过EVAL命令执行lua加锁
    List<RLock> acquiredLocks = new ArrayList<>(locks.size());
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        RLock lock = iterator.next();
        boolean lockAcquired;
        try {
            if (waitTime == -1 && leaseTime == -1) {
                lockAcquired = lock.tryLock();
            } else {
                long awaitTime = Math.min(lockWaitTime, remainTime);
                lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
            }
        } catch (RedisResponseTimeoutException e) {
            unlockInner(Arrays.asList(lock));
            lockAcquired = false;
        } catch (Exception e) {
            lockAcquired = false;
        }
        
        if (lockAcquired) {
            acquiredLocks.add(lock);
        } else {

            //计算已经申请锁失败的节点是否已经到达允许加锁失败节点个数限制 （N-(N/2+1)）
            if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
                break;
            }

            if (failedLocksLimit == 0) {
                unlockInner(acquiredLocks);
                if (waitTime == -1 && leaseTime == -1) {
                    return false;
                }
                failedLocksLimit = failedLocksLimit();
                acquiredLocks.clear();
                // reset iterator
                while (iterator.hasPrevious()) {
                    iterator.previous();
                }
            } else {
                failedLocksLimit--;
            }
        }

        // 计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false
        if (remainTime != -1) {
            remainTime -= System.currentTimeMillis() - time;
            time = System.currentTimeMillis();
            if (remainTime <= 0) {
                unlockInner(acquiredLocks);
                return false;
            }
        }
    }

    if (leaseTime != -1) {
        List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());
        for (RLock rLock : acquiredLocks) {
            RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
            futures.add(future);
        }
        
        for (RFuture<Boolean> rFuture : futures) {
            rFuture.syncUninterruptibly();
        }
    }
    
    return true;
}

可以看到其相关源码是非常好理解的。

总结

本文简单介绍了使用Redis实现分布式锁可能出现的问题，以及解决此问题的一种算法RedLock，并提供了简单的代码使用。

关于RedLock更多内容，可以查看官网 Distributed locks with Redis 这篇文章。

作者详细讨论了RedLock算法的安全性，可靠性等。