前期我在《【分布式】越不过去的分布式锁》一文中，提到过Redis实现分布式锁的常规思路，即基于SETNX的实现，里面提到Redis 官方站提出了一种权威的基于 Redis 实现分布式锁的方式名叫 Redlock，可弥补redis常规手段实现的天生缺陷。那么本篇，我们将对这一种更高级的分布式锁的实现方式Redlock进行实验并进一步探讨。
什么是RedlockRedis 官方站这篇文章提出了一种权威的基于 Redis 实现分布式锁的方式名叫 Redlock，此种方式比原先的单节点的方法更安全。它可以保证以下特性：
- 安全特性：互斥访问，即永远只有一个 client 能拿到锁
- 避免死锁：最终 client 都可能拿到锁，不会出现死锁的情况，即使原本锁住某资源的 client crash 了或者出现了网络分区
- 容错性：只要大部分 Redis 节点存活就可以正常提供服务
参见：
- https://redis.io/topics/distlock
- https://github.com/antirez/redis-doc/blob/master/topics/distlock.md
Redlock算法大概是这样的：
在Redis的分布式环境中，我们假设有N个Redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。我们确保将在N个实例上使用与在Redis单实例下相同方法获取和释放锁。现在我们假设有5个Redis master节点，分布在不同的机房尽量保证可用性。为了获得锁，client 会进行如下操作：
- 得到当前的时间，单位毫秒
- 尝试顺序地在 5 个实例上申请锁，当然需要使用相同的 key 和 random value，这里一个 client 需要合理设置与 master 节点沟通的 timeout 大小，避免长时间和一个 fail 了的节点浪费时间
- 当 client 在大于等于 3 个 master 上成功申请到锁的时候，且它会计算申请锁消耗了多少时间，这部分消耗的时间采用获得锁的当下时间减去第一步获得的时间戳得到，如果锁的持续时长（lock validity time）比流逝的时间多的话，那么锁就真正获取到了。
- 如果锁申请到了，那么锁真正的 lock validity time 应该是 origin（lock validity time） - 申请锁期间流逝的时间
- 如果 client 申请锁失败了，那么它就会在少部分申请成功锁的 master 节点上执行释放锁的操作，重置状态
当然，上面描述的只是获取锁的过程，而释放锁的过程比较简单：客户端向所有Redis节点发起释放锁的操作，不管这些节点当时在获取锁的时候成功与否。
分布式锁实现redisson已经有对Redlock算法封装，接下来对其用法进行简单介绍，并对核心源码进行分析（目前没有过多精力做redis集群，暂时使用单机模式，RedissonClient默认是支持单机，主从，哨兵，集群等模式的，可自定义配置）。
POM依赖
<!-- Redisson JDK 1.8+ compatible -->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.9.0</version>
</dependency>
基于springboot,RedissonClient自动装载配置：
@Configuration
@ConditionalOnClass(Config.class)
public class RedissonAutoConfiguration {
@Bean
public RedissonClient getRedisson() {
//支持单机，主从，哨兵，集群等模式
Config config = new Config();
config.useSingleServer()
.setAddress("redis://127.0.0.1:6377")
.setTimeout(2000)
.setConnectionPoolSize(50)
.setConnectionMinimumIdleSize(10);
RedissonClient redisson = Redisson.create(config);
try {
System.out.println("检测是否配置完成:"+redisson.getConfig().toJSON().toString());
} catch (IOException e) {
e.printStackTrace();
}
return redisson;
}
}
Redlock接口实现类，我们可以看到redission封装的redlock算法实现的分布式锁用法，非常简单，跟重入锁（ReentrantLock）有点类似：
/**
* Redlock 实现类
*/
@Component
public class RedissonDistributedLocker implements DistributedLocker {
@Autowired
private RedissonClient redissonClient;
/**
* 拿不到lock就不罢休，不然线程就一直block
* @param lockKey
* @return
*/
@Override
public RLock lock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock();
return lock;
}
/**
*
* @param lockKey
* @param timeout 加锁时间 单位为秒
* @return
*/
@Override
public RLock lock(String lockKey, long timeout) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, TimeUnit.SECONDS);
return lock;
}
/**
*
* @param lockKey
* @param unit 时间单位
* @param timeout 加锁时间
* @return
*/
@Override
public RLock lock(String lockKey, TimeUnit unit, long timeout) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, unit);
return lock;
}
/**
* tryLock()，马上返回，拿到lock就返回true，不然返回false。
* 带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false.
* @param lockKey
* @param unit
* @param waitTime
* @param leaseTime
* @return
*/
@Override
public boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime) {
RLock lock = redissonClient.getLock(lockKey);
try {
return lock.tryLock(waitTime, leaseTime, unit);
} catch (InterruptedException e) {
return false;
}
}
@Override
public void unlock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.unlock();
}
@Override
public void unlock(RLock lock) {
lock.unlock();
}
}
测试方法：
@Autowired
private DistributedLocker distributedLocker;
@Test
public void readLockTest() throws Exception {
String key = "redisson_key";
for (int i = 0; i < 100; i++) {
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
System.err.println("===线程开启===" + Thread.currentThread().getName());
/*
//直接加锁，获取不到锁则一直等待获取锁
distributedLocker.lock(key,10L);
//获得锁之后可以进行相应的处理
Thread.sleep(100);
System.err.println("===获得锁后进行相应的操作==="+Thread.currentThread().getName());
//解锁
distributedLocker.unlock(key);
System.err.println("==="+Thread.currentThread().getName());
*/
//尝试获取锁，等待5秒，自己获得锁后一直不解锁则10秒后自动解锁
boolean isGetLock = distributedLocker.tryLock(key, TimeUnit.SECONDS, 5L, 10L);
if (isGetLock) {
Thread.sleep(100); //获得锁之后可以进行相应的处理
System.err.println("===获得锁后进行相应的操作===" + Thread.currentThread().getName());
//distributedLocker.unlock(key);
System.err.println("===" + Thread.currentThread().getName());
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
t.start();
}
}
代码地址：
hi_leon/leon-ever-onwardgitee.com
注：由于换了新环境，github源码太大下载不下来，copy到码云了暂时。
源码分析 唯一ID实现分布式锁的一个非常重要的点就是set的value要具有唯一性，redisson的value是怎样保证value的唯一性呢？答案是UUID+threadId。入口在redissonClient.getLock("REDLOCK_KEY")，源码在Redisson.java和RedissonLock.java中：
final UUID id;
protected String getLockName(long threadId) {
return this.id + ":" + threadId;
}
获取锁的代码为redLock.tryLock()或者redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS)，两者的最终核心源码都是下面这段代码，只不过前者获取锁的默认租约时间（leaseTime）是LOCK_EXPIRATION_INTERVAL_SECONDS，即30s：
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
// 获取锁时需要在redis实例上执行的lua命令
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
// 首先分布式锁的KEY不能存在，如果确实不存在，那么执行hset命令（hset REDLOCK_KEY uuid+threadId 1），并通过pexpire设置失效时间（也是锁的租约时间）
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
// 如果分布式锁的KEY已经存在，并且value也匹配，表示是当前线程持有的锁，那么重入次数加1，并且设置失效时间
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
// 获取分布式锁的KEY的失效时间毫秒数
"return redis.call('pttl', KEYS[1]);",
// 这三个参数分别对应KEYS[1]，ARGV[1]和ARGV[2]
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
获取锁的命令中，
- KEYS[1]就是Collections.singletonList(getName())，表示分布式锁的key，即REDLOCK_KEY；
- ARGV[1]就是internalLockLeaseTime，即锁的租约时间，默认30s；
- ARGV[2]就是getLockName(threadId)，是获取锁时set的唯一值，即UUID+threadId：
释放锁的代码为redLock.unlock()，核心源码如下：
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
// 释放锁时需要在redis实例上执行的lua命令
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
// 如果分布式锁KEY不存在，那么向channel发布一条消息
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
// 如果分布式锁存在，但是value不匹配，表示锁已经被占用，那么直接返回
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
// 如果就是当前线程占有分布式锁，那么将重入次数减1
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
// 重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只设置失效时间，还不能删除
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
// 重入次数减1后的值如果为0，表示分布式锁只获取过1次，那么删除这个KEY，并发布解锁消息
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
// 这5个参数分别对应KEYS[1]，KEYS[2]，ARGV[1]，ARGV[2]和ARGV[3]
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));
}
由于N个Redis节点中的大多数能正常工作就能保证Redlock正常工作，因此理论上它的可用性更高。我们前期讨论的单Redis节点的分布式锁在failover的时候锁失效的问题，在Redlock中不存在了（解决了遗留问题１），但如果有节点发生崩溃重启，还是会对锁的安全性有影响的。具体的影响程度跟Redis对数据的持久化程度有关。
根据上述提出的算法，当Ｎ个节点中有一个节点宕机，仍然存在锁的安全性问题。具体的影响跟redis的持久化程度有关。
假设一共有5个Redis节点：A, B, C, D, E。设想发生了如下的事件序列：
- 客户端1成功锁住了A, B, C，获取锁成功（但D和E没有锁住）。
- 节点C崩溃重启了，但客户端1在C上加的锁没有持久化下来，丢失了。
- 节点C重启后，客户端2锁住了C, D, E，获取锁成功。
这样，客户端1和客户端2同时获得了锁（针对同一资源）。
在默认情况下，Redis的AOF持久化方式是每秒写一次磁盘（即执行fsync），因此最坏情况下可能丢失1秒的数据。为了尽可能不丢数据，Redis允许设置成每次修改数据都进行fsync，但这会降低性能。当然，即使执行了fsync也仍然有可能丢失数据（这取决于系统而不是Redis的实现）。所以，上面分析的由于节点重启引发的锁失效问题，总是有可能出现的。为了应对这一问题，antirez又提出了延迟重启(delayed restarts)的概念。也就是说，一个节点崩溃后，先不立即重启它，而是等待一段时间再重启，这段时间应该大于锁的有效时间(lock validity time)。这样的话，这个节点在重启前所参与的锁都会过期，它在重启后就不会对现有的锁造成影响。
关于Redlock还有一点细节值得拿出来分析一下：
在最后释放锁的时候，antirez在算法描述中特别强调，客户端应该向所有Redis节点发起释放锁的操作。也就是说，即使当时向某个节点获取锁没有成功，在释放锁的时候也不应该漏掉这个节点。
这是为什么呢？设想这样一种情况，客户端发给某个Redis节点的获取锁的请求成功到达了该Redis节点，这个节点也成功执行了SET操作，但是它返回给客户端的响应包却丢失了。这在客户端看来，获取锁的请求由于超时而失败了，但在Redis这边看来，加锁已经成功了。因此，释放锁的时候，客户端也应该对当时获取锁失败的那些Redis节点同样发起请求。实际上，这种情况在异步通信模型中是有可能发生的：客户端向服务器通信是正常的，但反方向却是有问题的。
其它问题１、仍然存在客户端长时间阻塞，导致获得的锁释放，访问的共享资源不受保护的问题。
２、在Redlock的算法中，我们可以看到第３步，当获取锁耗时太多，留给客户端的访问共享资源的时间很短，这种情况若来不及操作，是不是要释放锁呢？且到底剩下多少时间才算短？这又是一个选择难题。
３、Redlock算法对时钟依赖性太强，若Ｎ个节点中的某个节点发生时间跳跃，也可能会引此而引发锁安全性问题。
结束语关于分布式锁，先告一段落，最近过于忙碌，新公司的技术栈又过于老旧，dubbo、spring2.x、struts2、JDBC、MongoDB、Redis、memecache...整个一个大杂烩，我的这两周的心情犹如万只羊驼在奔腾。只能期待未来的日子，在技术选型上，可以有发挥的余地。
三十而立的年纪，修炼成佛！
参考：
- https://redis.io/topics/distlock
- redis分布式锁的安全性探讨（二）：分布式锁Redlock https://blog.csdn.net/hh1sdfsf56456/article/details/79474434
- Redlock：Redis分布式锁最牛逼的实现https://www.jianshu.com/p/7e47a4503b87
【分布式】基于Redis实现分布式事务锁mp.weixin.qq.com