一文搞懂Redis分布式锁
前言：
日常开发中，我们经常会使用到锁，以保证某一段逻辑是线程安全的，同步的。 但是当今一般都是同一个服务部署到多台机器上，在这种情况下，如果用java中的锁，将只能保证在某台机器上的线程安全，而不能保证真正意义上的线程安全，那么此时分布式锁就上场了。
下边我们将以层层递进的方式，看看怎么用redis实现 相对完美的 分布式锁！
- 多说一句：一把合格的分布式锁，至少应该保证以下特性：
1:互斥性 (保证锁内代码逻辑同步) 2:超时自动释放（防止死锁，占用资源） 3:安全性（不能删除非自己加的锁） 4:原子（放值和取值时保证原子） 5:高可用，高性能 6:支持可重入 7:支持自动续期
1. setnx + expire 实现一个简陋的锁
setnx 命令解释：如果数据库不存在给定的key， 则设置成功返回true，如果存在则设置不成功，返回false，
值的注意的是在一些极端情况下（比如锁中代码执行时间过长，或者没有成功删除锁对应的key,此时将会一直占用资源（也就是死锁），其他线程获取不到锁，会出现很严重的问题）？所以我们给这个key加个过期时间以便给这个key加一个”约束”，使得资源及时释放，伪代码如下：
if ( setnx( key1, value1)==1 ){
expire(key1);
try {
//TODO 执行业务代码...
} finally {
del(key);
}
}else{
//未获取到锁
}
1.1 setnx与expire非原子问题以及解决
but这样存在一个问题（因为setnx和expire不是原子操作），也就代表，在某一时刻，可能setx执行了，但是expire没执行。此时就又有可能出现死锁问题。在业界用的最多的一种就是使用lua脚本解决该问题，即通过lua将setnx和expire两个命令变成原子操作。如下：
if (redis.call('setnx',KEYS[1],ARGV[1]) < 1) then return 0; end; redis.call('expire',KEYS[1],tonumber(ARGV[2])); return 1;" 1 key value 100
从上脚本可以看出，调用setnx如果设置成功则紧接着调用expire命令设置过期时间。
1.2 B锁 被A删情况以及解决
Ok,上边的lua脚本是保证了原子，但是删除时候又出现问题了。
-
考虑如下场景：
线程A 抢到锁，执行业务逻辑，但是超过了锁的过期时间（比如：10秒）也没执行完，此时由于到了过期时间，所以redis中会删除该key,此时线程B恰好过来抢到了锁，在执行过程中，线程A 执行完了，此时就会进入finally中执行del(key)，于是线程A把线程B加的锁给删除了。。。
实时上，A删B锁这种场景不仅是加锁解锁逻辑的混乱，还会导致并发，即线程A删除B加的锁后，设想此时线程C过来并且抢到锁，那么线程c也会执行锁中代码快（假设此时线程B还没结束），也就是说有线程B 线程C并发执行锁中代码，此时锁已经不起作用，形同虚设。。。（士可忍孰不可忍）
为了避免这种严重问题，我们的解决方式是：在某个线程去设置锁的时候，value给一个uuid,当删除key时候，我们先get出来然后和设置时候比对一下，如果等于那就说明是当前线程设置的锁，执行del,否则就说明不是，则不进行del,从而避免A锁B删这种现象发生。
but此时又有个问题，就是解锁时候get操作和del操作不是原子的，那么我们还可以使用lua来解决这个问题，如下：
if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end
ok到这里是不是感觉没问题了？答案是：no！
在实际场景中，我们可能对分布式锁有其他的要求，比如一些特殊场景下需要支持：
可重入：（即线程a拿到锁后，在持有锁期间，再次去申请这把锁是可以申请成功的（个人认为这种情况很少遇到，我能想到的有：1：递归逻辑。2：锁中代码逻辑再次请求和外部同key的锁（ps:真实场景似乎没遇到，我们这里仅仅讨论理论上的可能性） ））自动续期：在某些场景下，需要保证锁超时后，要自动续期，以保证没执行完的逻辑继续执行。
等特性，我们这里不再班门弄斧，直接来看下业内大拿：Redisson是怎么做的。
2. redisson锁实现
作为redis客户端框架 redisson不止提供了开箱即用的分布式锁api,还包括其他很多特性这里不再展开，更多请关注：redisson官网 ，我们只是关注redisson是如何实现的分布式锁的。
首先我们要知道的是 redisson实现的分布式锁api,基本上是一把合格的锁，他保证了我们开篇说的那几个特性，如下：
1:互斥性 (保证锁内代码逻辑同步)
2:超时自动释放（防止死锁，占用资源）
3:安全性（不能删除非自己加的锁）
4:原子（放值和取值时保证原子）
5:高可用，高性能
6:支持可重入
7:支持自动续期
要想分析redisson如何实现，那必然离不了源码（注意：这里我们只是简单过一遍加锁，解锁，重入性，自动续期等实现逻辑,其他略过）
2.1 加锁
废话不多说，我们直接来到加锁脚本这，如下：
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " + //先检查下给定的key 存不存在，不存在进入下边分支逻辑
//如果不存在，那么往redis写入一个hash结构的数据(hincrby命令如果检测到没有key时，会写入)，
//其中key是调用者传入的key,value是map类型，其中key是客户端id(看源码的话知道他是一个uuid每一个连接都保持唯一):threid ，
//value就是某个线程的获取锁的次数（这个操作是实现可重入的保障）
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +//给刚刚设置的key 设置过期时间
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + //如果给定的key存在,
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +//将该key中对应的value值的value值 +1 操作，以记录某个key下 某个线程的重入次数
"redis.call('pexpire', KEYS[1], ARGV[1]); " +//重入后，更新过期时间为给定值
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);", //没获取到锁，则返回锁的剩余时间
Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
protected String getLockName(long threadId) {
return id + ":" + threadId;//这里的id是CommandAsyncExecutor实现类中的id,本质上和某一条连接一一对应，值是uuid
}
读完上边脚本代码和注释，我们知道redisson
- 通过lua脚本来保证多命令原子性，
- 通过hash结构来存储锁信息
锁 结构如下： { 调用方给定的key : {redisson客户端id（是个uuid，每个连接唯一）+获取到锁的线程id : 该线程获取到的锁次数 } } - 通过判断当前线程是否持有锁，持有的话 value值+1，来记录某个线程获取锁的次数，从而间接实现了
重入的特性
一张图直观看下加锁流程：
2.2 解锁
废话不多说直接上代码
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +//不存在key 则直接返回，不做操作
"return nil;" +
"end; " +
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +//否则根据key和 当前连接id+当前线程id 将value值减一
"if (counter > 0) then " +//如果减一后value大于0 说明该线程重入过，
"redis.call('pexpire', KEYS[1], ARGV[2]); " +//将该线程持有的锁的过期时间更新
"return 0; " +
"else " +//counter不大于0，则删除key 释放锁
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +//发布事件，广播给其他订阅该事件的线程，通知他们，”可以尝试抢锁啦！“
"return 1; " +//返回释放锁成功
"end; " +
"return nil;",
Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}
从上边可以看到:
- 解锁时，如果重入过n次，那么就需要释放n次，直到value值为0 才真正删除 key。
- 另外 在删除时候，我们可以看到先判断存在与否，而这里的判断是根据给定key和连接id+线程id来的，也就是说，不会发生B锁A删的情况出现。
- 同时在删除锁后，发布事件，告知其他等待线程可以抢锁啦！
2.3 自动续期
我们来到 tryAcquireAsync 方法，如下：
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
RFuture<Long> ttlRemainingFuture;
if (leaseTime != -1) {
ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
}
//加锁成功
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e != null) {
return;
}
// lock acquired
if (ttlRemaining == null) {
if (leaseTime != -1) {
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
//可以看到如果leaseTime == -1 （也就是没有传leaseTime参数）那么就会自动续期
scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
续期代码：
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +//该线程持有锁，
"redis.call('pexpire', KEYS[1], ARGV[1]); " +// 重置过期时间为 internalLockLeaseTime 即 30秒（redisson代码中默认的，但可以配置）
"return 1; " +
"end; " +
"return 0;",
Collections.singletonList(getRawName()),
internalLockLeaseTime, getLockName(threadId));
}
自动续期内部源码不在展开，我们简单介绍下主要流程：
只要线程加锁成功并且 没传leaseTime值（leaseTime我喜欢叫它租约时长，这个参数很重要！！！，在使用时候一定要注意，如果调用tryLock时，指定了租约时长，那么redisson看门狗机制将不起作用）此时该值会默认为 -1 ，就会启动一个timer线程（也有人称为watch dog看门狗机制），它是一个定时任务（基于netty的Timer类实现的），会每隔10（internalLockLeaseTime（默认为30）/3 =10 ）秒检查一下，如果线程A还持有锁，那么就会不断的延长锁key的生存时间。从而实现了 自动续期。
注意：一旦你开启了自动续期（watch dog机制），那么一定要合理控制锁中逻辑的执行时间，避免执行时间过长（事实上，大多数的逻辑我们都应该尽可能的快速执行完毕）。
ok到这里我们用一张图，来直观看下redisson整体的一个流程：
2.4 集群情况下存在的问题以及RedLock
到这redisson我们就说完了，但是有人会说，使用redisson这个分布式锁肯定就能保证完美无瑕吗？
- 单机版 redisson 已经 yyds ，很完美了至少我个人觉得
- 集群 redisson 还是有问题，锁并不一定安全，不能真正保证互斥
考虑如下场景：
如果线程A在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时， master节点发生故障，一个slave节点就会升级为master节点。线程B就可以获取同个key的锁啦，但线程A 也已经拿到锁了，此时两个不同线程都拿到锁，并发执行锁中代码逻辑，锁的安全性，互斥性也就荡然无存了！
基于上述集群存在的问题，Redis作者 antirez 大佬提出一种高级的分布式锁算法：Redlock。核心思想如下：
不能只在一个redis实例上创建锁，应该是在多个redis实例上创建锁(
意味着那就要部署多个master)，并且必须在 (n/2)+1 个master节点上都成功创建锁，才能算这个整体的RedLock加锁成功，避免说仅仅在一个redis实例上加锁，然后同步到slave时带来的问题。
而redisson是实现了RedLock了的，其实现也很简单，即遍历所有的Redis客户端，然后依次加锁（也得看redis集群是什么模式的 哨兵应该是不支持RedLock的，因为slave不能由客户端直接写），最后统计成功的次数来判断是否加锁成功。
public class RedissonMultiLock implements RLock
该类RedissonMultiLock即是RedLock的实现。具体如何使用以及有哪些坑我们不去展开，（ps:个人感觉RedLock的思想有点投入与产出不成正比，如果业务一定保证稳定性，互斥，安全，且并发不算太高时候，也可以用zk实现的分布式锁，没必要这么费时费力，还得搭建多个master）
到此redis分布式锁就写完了，我们一般工作中也不造轮子，直接用redisson就好了，但是一定要去了解里边的实现，否则很容易踩坑呀！