Redisson实现Redis分布式锁

在上篇文章redis实现分布式锁中，讲了通过redis实现分布式锁的正确方式，并在文章最后讲到，文章中的实现方式存在单点问题，并提到一种基于分布式环境的分布式锁——Redlock。本文就讲解一下Redlock的一些概念和实现。

1. Redisson

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish/Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service。Redisson提供了使用Redis的最简单和最便捷的方法。

Redisson是一个开源项目，Github地址：https://github.com/redisson/redisson

在Redisson中，提供了Redis多种分布式锁的实现（包括Redlock），下面首先来看一下通过Redisson如何实现一个普通的分布式锁。

1.1 Redis架构

Redis在实际使用中，存在以下几种常见的部署架构：

1. 单机模式
2. 主从模式
3. 哨兵模式
4. 集群模式

对于这几种Redis部署架构，Redisson都是支持的。

1.2 Redisson实现普通分布式锁

之前那篇文章redis实现分布式锁，介绍了通过Jedis客户端实现普通分布式的正确姿势，接下来介绍一下通过Redisson如何在上述几种Redis部署架构下实现普通分布式锁，其实Redisson实现普通分布式锁的原理跟上篇文章介绍的完全一致，都是Redis通过EVAL命令执行LUA脚本实现的。

1.2.1 单机模式

//redisson Config
Config config = new Config();
config.useSingleServer().setAddress("myredisserver:6379");

//构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);

//设置锁key
RLock disLock = redissonClient.getLock("DISLOCK");

try {
    //尝试获取分布式锁
    boolean isLock = disLock.tryLock(500, 15000, TimeUnit.MILLISECONDS);
    if (isLock) {
        //TODO if get lock success, do something;
        Thread.sleep(15000);
    }
} catch (Exception e) {
    //异常处理
} finally {
    //解锁
    disLock.unlock();
}

1.2.2 主从模式

实现代码和单机模式几乎一样，唯一的不同就是Config的构造：

Config config = new Config();
config.useMasterSlaveServers()
    //可以用"rediss://"来启用SSL连接
    .setMasterAddress("redis://127.0.0.1:6379")
    .addSlaveAddress("redis://127.0.0.1:6389", "redis://127.0.0.1:6332", "redis://127.0.0.1:6419")
    .addSlaveAddress("redis://127.0.0.1:6399");

1.2.3 哨兵模式

Config config = new Config();
config.useSentinelServers()
    .setMasterName("mymaster")
    //可以用"rediss://"来启用SSL连接
    .addSentinelAddress("127.0.0.1:26389", "127.0.0.1:26379")
    .addSentinelAddress("127.0.0.1:26319");

1.2.4 集群模式

Config config = new Config();
config.useClusterServers()
    .setScanInterval(2000) // 集群状态扫描间隔时间，单位是毫秒
    //可以用"rediss://"来启用SSL连接
    .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001")
    .addNodeAddress("redis://127.0.0.1:7002");

跟Jedis自己实现的分布式锁相比，通过Redisson实现普通分布式锁很简单，支持各种Redis的部署方式，也支持公平锁、非公平锁等。其他配置详见：Redisson Wiki-配置

2. Redlock分布式锁

2.1 Redlock概念

Redis作者antirez基于分布式环境下提出分布式锁实现方式Redlock算法大致如下：

在Redis的分布式环境中，我们假设有N个Redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。我们确保将在N个实例上使用与在Redis单实例下相同方法获取和释放锁。现在我们假设有5个Redis master节点，同时我们需要在5台服务器上面运行这些Redis实例，这样保证他们不会同时都宕掉。

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

1. 获取当前Unix时间，以毫秒为单位
2. 依次尝试从5个实例，使用相同的key和具有唯一性的value（例如UUID）获取锁。当向Redis请求获取锁时，客户端应该设置一个获取锁的超时时间，这个超时时间应该小于锁的失效时间。比如获取锁的超时时间是50毫秒，锁自动失效时间为3秒。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，则会重新计算获取锁的剩余时间（超时时间 – 获取），并尽快尝试去另外一个Redis实例请求获取锁
3. 如果步骤2获取锁成功，则重新计算接下来获取锁剩余的超时时间（超时时间 – 获取锁已使用的时间），如果剩余超时时间大于0，则会尽快去获取下一个锁
4. 如果步骤2获取锁失败，则会判断当前已经成功获取的锁的个数是否大于等于N/2 + 1，如果满足条件，说明已经获取到大部分锁，表示竞争Redlock成功，不会再去竞争其他锁；否则判断当前已获取失败的锁的个数是否达到(N – (N/2 + 1))，如果以达到个数，则表示竞争Redlock失败。否则，重新计算获取剩余的超时时间，并去竞争下一个Redis实例的锁

2.2 Redisson实现

Redisson实现了Redlock，那么具体怎么使用呢？下面通过单机模式Redis为例，看一下如何使用Redlock，如下：

Config config = new Config();
config.useClusterServers().addNodeAddress(
        "redis://127.0.0.1:6379","redis://127.0.0.1:6369", "redis://127.0.0.1:6359",
        "redis://127.0.0.1:6349","redis://127.0.0.1:6339")
        .setPassword("******");

Config config1 = new Config();
config1.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redissonClient1 = Redisson.create(config1);

Config config2 = new Config();
config2.useSingleServer().setAddress("redis://127.0.0.1:6378");
RedissonClient redissonClient2 = Redisson.create(config2);

Config config3 = new Config();
config3.useSingleServer().setAddress("redis://127.0.0.1:6377");
RedissonClient redissonClient3 = Redisson.create(config3);

String resourceName = "REDLOCK";
RLock lock1 = redissonClient1.getLock(resourceName);
RLock lock2 = redissonClient2.getLock(resourceName);
RLock lock3 = redissonClient3.getLock(resourceName);

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
try {
    boolean isLock = redLock.tryLock(500, 30000, TimeUnit.MILLISECONDS);
    System.out.println("isLock = "+isLock);
    if (isLock) {
        //TODO if get lock success, do something;
        Thread.sleep(30000);
    }
} catch (Exception e) {
} finally {
    //解锁
    redLock.unlock();
}

跟Redisson实现普通分布式锁一样，如果是主从Redis架构、哨兵Redis架构、集群Redis架构实现Redlock，只需要改变上述config1、config2、config3为主从模式、哨兵模式、集群模式配置即可，但相应需要3个独立的Redis主从集群、3个Redis独立的哨兵集群、3个独立的Cluster集群。

2.3 源码分析

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
    long newLeaseTime = -1;
    if (leaseTime != -1) {
        newLeaseTime = waitTime*2;
    }
    
    long time = System.currentTimeMillis();
    long remainTime = -1;
    if (waitTime != -1) {
        remainTime = unit.toMillis(waitTime);
    }

    //获取允许加锁失败节点个数(N-(N/2+1))
    int failedLocksLimit = failedLocksLimit();

    //用于存储加锁成功的RLock
    List<RLock> lockedLocks = new ArrayList<RLock>(locks.size());

    //遍历所有的锁
    for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
        RLock lock = iterator.next();
        boolean lockAcquired;
        try {
            if (waitTime == -1 && leaseTime == -1) {
                //每个redis节点加锁不设超时时间
                lockAcquired = lock.tryLock();
            } else {
                //等待每个Redis节点加锁的时间是waitTime，自动解锁时间是waitTime * 2
                long awaitTime = unit.convert(remainTime, TimeUnit.MILLISECONDS);
                lockAcquired = lock.tryLock(awaitTime, newLeaseTime, unit);
            }
        } catch (Exception e) {
            lockAcquired = false;
        }
        
        if (lockAcquired) {
            //加锁成功，讲当前锁加入lockedLocks
            lockedLocks.add(lock);
        } else {
            //加锁失败
            if (locks.size() - lockedLocks.size() == failedLocksLimit()) {
                //已成功获取(N/2+1)个节点的锁，直接跳出循环
                break;
            }

            if (failedLocksLimit == 0) {
                //已经达到允许失败的最大个数，对所有已加的锁进行解锁
                unlockInner(lockedLocks);
                if (waitTime == -1 && leaseTime == -1) {
                    //竞争Redlock失败
                    return false;
                }
                failedLocksLimit = failedLocksLimit();
                lockedLocks.clear();
                // reset iterator
                while (iterator.hasPrevious()) {
                    iterator.previous();
                }
            } else {
                failedLocksLimit--;
            }
        }
        
        //重新计算获取锁的剩余时间
        if (remainTime != -1) {
            remainTime -= (System.currentTimeMillis() - time);
            time = System.currentTimeMillis();
            if (remainTime <= 0) {
                //时间已用完，释放之前竞争到的节点的锁，返回false，竞争Redlock失败
                unlockInner(lockedLocks);
                return false;
            }
        }
    }

    //已成功获取大于等于(N/2+1)个节点的锁，则给所有已获取到的节点的锁设置一个超时时间leaseTime，时间到了会自动解锁
    if (leaseTime != -1) {
        List<RFuture<Boolean>> futures = new ArrayList<RFuture<Boolean>>(lockedLocks.size());
        for (RLock rLock : lockedLocks) {
            RFuture<Boolean> future = rLock.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
            futures.add(future);
        }
        
        for (RFuture<Boolean> rFuture : futures) {
            rFuture.syncUninterruptibly();
        }
    }
    
    //竞争Redlock成功
    return true;
}

以sentinel模式架构为例，假如有sentinel-1，sentinel-2，sentinel-3总计3个sentinel模式集群，如果要获取Redlock分布式锁，那么需要向这3个sentinel集群通过EVAL命令执行LUA脚本，需要至少2个sentinel集群响应成功，才算成功的以Redlock算法获取到Redlock分布式锁。

可以看到，虽然Redisson为我们封装了Redlock的实现，代码层面也比较简单，但是由于Redlock的特性，要使用多个（3个及以上）集群，代价是比较大的。如果仅仅是追求高可用，完全可以使用Zookeeper来实现分布式锁，但是从效率上看，肯定是Redis更高效的，至于如何选择，还要看具体业务场景。

参考链接：

1. Redisson Wiki – Redisson配置

2. Redis中文社区——Redis分布式锁

3. Redlock：Redis分布式锁最牛逼的实现