【Redis笔记】一起学习Redis | 如何利用Redis实现一个分布式锁？

一起学习Redis | 如何利用Redis实现一个分布式锁？

前提知识
- 什么是分布式锁？
- 为什么需要分布式锁？
- 分布式锁的5要素和三种实现方式
实现分布式锁
- 思考思考
- 基础方案
- 改进方案
- 保证setnx和expire的复合操作的原子性
依然可能存在的不足

前提知识

什么是分布式锁？

分布式锁简而言之就是集群环境下，全局意义上的锁，会对集群上的所有节点服务都造成关联和影响

为什么需要分布式锁？

我们都知道，当有场景需要解决代码的并发安全问题的时候，我们都会给相应的代码端或方法加锁，无论你是使用Synchronized内置锁还是显示锁Reentrantlock等其他方式实现；但是这种加锁的方式仅仅局限于单机的情况下，在服务集群的环境下，可能前面的加锁方式就显得有心无力了

比如说我有一个服务A，它由3个集群节点组成，在服务A的某个方法上需要顺序串行执行，所以在这个方式加了锁；当上游服务来了几千个并发请求同时要求执行服务A的这个方法，此时微服务的负载均衡会将这几千个请求分散到这个服务A的3个集群节点中，虽然我们加了锁，但同一时刻，因为集群环境的关系，至少有3个线程在执行这个加了锁的方法，从而造成一定的线程安全点。

因为我们加的锁仅仅是一个单机锁，而非去全局下的一个锁，那么要实现一个全局意义上的锁，我们应该怎么实现呢？ 首先这在JDK本身的代码中，肯定是没有现成的实现的。

分布式锁的5要素和三种实现方式

分布式锁的五点要素：

保证在分布式集群环境中，同一个方法在同一时间只能被其中的一个节点的一个线程执行

为避免死锁，这把锁要是一把可重入锁

这个锁最好是一把阻塞锁，如果有需要，要让其他没获得锁的线程阻塞住

有高可用的获取锁和释放锁功能，即要有一定的可靠性

获取锁和释放锁的性能要好

以上五点非常的重要，每当我们要设计一个分布式锁方案的时候，我们就要认真的考虑我们的方案是否满足这5点，当然如果你的需求不需要满足这么多要求，也可以不满足

分布式锁一般有三种实现方式：

数据库锁

基于Redis的分布式锁

基于ZooKeeper的分布式锁

在我查到的中文资料中，大部分常见的分布式锁方案都是通过这三种途径去实现的，比如利用数据库的主键唯一索引特性，有数据库记录代表有锁，释放锁则删除该记录；又或者利用数据库的悲观锁机制等待；但是通常情况下数据库的方式运用的比较少，毕竟首先就存在性能问题，所以这个可以在小规模的场景玩一下

从理解的难易程度角度（从低到高）: 数据库 > 缓存 > Zookeeper
从实现的复杂性角度（从低到高）: Zookeeper >= 缓存 > 数据库
从性能角度（从高到低）: 缓存 > Zookeeper >= 数据库
从可靠性角度（从高到低）: Zookeeper > 缓存 > 数据库

本博在生产环境中呢，暂且只使用过Redis的方案，所以也主要介绍的是如果通过Redis去实现一个分布式锁，所以呢，其他方式就不多说啦

实现一个分布式锁

思考思考

我先声明一下，这里要设计的分布式锁是符合前面说道的五要素的，即尽量的保证在功能上接近原生的锁(Synchronized),所以我们必须保证同一时刻只有一个节点的一个线程获的锁，是可重入锁，可以实现阻塞，有一定可靠性，获得锁和释放锁的性能高

首先，Redis可以做集群多节点，肯定是可以保证高可用的，同时Redis本身就是做缓存的，也可以保证锁的性能，所以这两点是Redis天然可以保障的；其他的三要素则是要通过代码层面的逻辑去实现的，那我们怎么去实现呢？

同一时间只有一个节点的线程获得锁
可阻塞
可重入

一、首先搞定同一时间只有一个节点获得锁

这一步必然不可少的就是Redis的setnx命令, 该命令的好处是有返回值，我们可以知道该命令到底有没有插入数据成功；即当key不存在的时候，set入Value, 存在则set失败；所以setnx的特点就是有返回值，成功插入返回1，已存在则返回0；业务意义就是返回1则代表获取到锁，Redis还没有其他线程获取到这锁，返回0 则代表已经有其他线程获取到锁了

我们的锁在Redis中存储结构通常是key-value结构，key是固定的，value可以是一个全局唯一值 ,可以随机数生成，在分布式集群环境建议使用全局唯一Id生成器生成，比如Snowflake算法

为了防止因为某个线程长时间不释放锁到导致死锁，我们还要给锁一个超时时间，可以通过expire(key , timeout)命令去实现

第一步： setnx(key , 全局唯一值) , 超时时间比如我们定义5s ，即5000ms; 返回1则成功获取锁，返回0则不成功要重试
第二步： expire(key , 5000) , 给锁一个超时时间，即使线程超时不释放锁，Redis也会根据key是否过期自动释放
第三步： 因为已经获取了锁，所以就可以执行真正的业务
第四步： del(key) , 业务执行成功，释放锁

行，同一时间只有一个节点获得锁的问题解决了，那我们就来解决一下可重入性的问题

二、然后我们要解决锁的可重入性

这个可以有多种实现方式，但是我个人觉得通过ThreadLocal的方式最简单。当我的线程在某个加锁方法中获取到了锁，我就给该线程的ThreadLocal设置一个flag ，比如true。但我在这个加锁方法里又调用了另一个加锁方法，我可以在这个线程的ThreadLocal中获得flag看看是否已经获得过锁，如果获的过，那我就可以什么都不做，继续执行。如果没有获得过，我就要再次重新获得锁

三、然后我们要解决锁的可阻塞性

什么叫可阻塞性呢？意思就是当某个节点的某个线程获得了锁，其余的所有线程在我释放锁前必须阻塞起来。 其实这个就看你怎么设计了，有的场景不一定需要阻塞，比如说Spring Scheduled + 分布式锁的任务调度场景，因为我只是想隔多久执行一个定时任务，但我不需要每个节点都启动一起，所以就拿分布式锁来保证在某个时间点上只需要一个节点来执行定时任务即可，获得锁的就执行定时任务，没有获得锁的就直接结束任务即可

当然有的场景，分布式锁还需要完全的模拟出锁的特性，即阻塞其他线程；此时我们只需要让没有获得锁的其他线程while循环获取锁，如果觉得频繁，可以给一个随机的休眠时间再重试(但是不要太长，毫秒级即可)；如果你觉得不能无限重试，就可以额定一个重试次数限制

基础的方案

行，上面我们解决了锁的可重入性，可阻塞性，同时间只有一个线程获得锁的特性，再加上Redis天然的可以通过集群的方式支持高可用性以及本身缓存就具有非常高的性能, 这样我们就毫无压力的把5大要素就实现了；

嗯嗯，非常完美~ 看起来毫无违和感，的确是发现不出有什么毛病

首先判断是否已经获得过锁，如果没有则生成Value，执行setnx命令尝试获取锁

如果成功获得锁，则给锁一个过期时间，同时开始执行正在业务，结束后释放锁，del（key）

如果没有成功获得锁，则while循环重试，可以给一个阻塞时间，时隔多久重试一次，也可以限制重试次数

缺陷：

但是，有时候，只要存在多方服务通过网络进行通讯，就避免不了出现一些网络故障的意外或程序意外

比如当某个线程的setnx命令与redis服务端交互成功后，既成功获取锁后；准备执行expire命令时，网络中断了，或者程序崩溃了，从而导致某个线程获取了锁，但没有给该锁设置一个过期时间。当网络恢复时，或程序重新上线时，就会发现其他线程一直被阻塞，一直无法获的锁，从而导致出现死锁问题

这个要怎么解决呢？所以我们来看一下改进后的方案

通过补偿机制去改进方案

当出现了setnx和expire之间的程序或网络中断之后，我们要怎么通过一个补偿机制去抵消可能出现的死锁呢？

问题：

首先根本问题是Redis中可能存在没有过期时间的锁，那么我们要解决的就是怎么让这个锁有过期时间，既然expire的方式不能完全保证过期时间的设置，那我们想想能不能在其他方式上实现这个过期时间呢？

当然是有的，我们之前谈到，在Redis中存储结构通常是key-value结构，key是固定的，value是一个全局唯一值 ；但在这里我们要将value改进一下，不再使用一个没有业务含义的唯一值做为Value。所以我们这里建议的Value 是当前时间戳 + 超时时间, 如下

key	value	备注
your_key	时间戳 + 超时时间	数值相加
my_lock_example	142435559000	时间戳142435554000ms，超时时间是5000ms

那么我们解决方案就是：

将锁的Value由全局随机唯一值，改为由当前时间戳 + 超时时间的格式 , 这个Value的业务实际意义就是锁的超时时的时间戳，既记录着该锁超时时的时间点

既然我们无法保证Redis去让它过期，那我们就让线程尝试获取锁失败的时候，再去取锁的Value，既锁的超时实际时间，用当前时间的时间戳和Value的时间戳去比较，看当前时间是否大于锁的超时时刻的时间，如果大于则代表该锁实际已经超时了，可以被其他线程重新获取，如果小于，则进入重试获取锁的阶段

这样人为的增加一次锁实际是否超时的判断，就可以弥补Redis设置过期时间失效的情况。当然相应的缺点就是多了一步网络请求操作，但也是值得的

另外通过这种方式去实现，你还要去解决一个场景，就是多个线程都获取锁失败了，都进入了get(value)与当前时间判断的代码段，如果他们都发现锁已超时，都想去替换锁的时候，你就要去想如何避免多个线程都获取到锁的问题了，当然最简单的方式就是给这个块代码段加锁，问题是单机锁无法实现集群的环境的锁，若又引入一个分布式锁，那更是让情况复杂化；所以这里通过会用getset命令去实现一个全局意义上的CAS操作,具体可以看下面的流程图

完整改进型方案的流程图：

当然画的不好请见谅，毕竟原本就是画给自己看看…

其实也不复杂，相比上一次的方案流程图，也就是多了线程获取锁失败后要增加的部分操作罢了

如果获取锁失败，则获取锁的Value，既锁的实际超时时间；用当前时间与锁超时时间比较

如果当前时间大于锁的超时时间或锁的Value为null(代表该key被其他线程删除了)，则代表该锁可以被替换了

但是为了解决并发情况下，多个线程都获取到锁的情况，所以我们要再做一个判断。既用新锁替换旧锁的同时获得旧锁的Value, 拿这个Value与上一步获得的Value进行比较，看是否一致，如果一致则代表，则是没有其他线程在跟自己竞争，则可以正式的获得锁；如果不一致，则代表我替换旧锁的时候，也有其他线程也替换了旧锁。其实本质上就是一个CAS操作

这种方式的实现，使用Spring Data Redis就可以很好的实现了，用其他的客户端方式当然也行

这里提供一个Java代码的实现, 基于Spring Data Redis实现

@Component
public class RemoteLock {private static final String LOCK_KEY_PREFIX = "LOCK-";private static ThreadLocal<AtomicInteger> threadLocal = new ThreadLocal<>();@Autowiredprivate RedisTemplate<String, Object> redisTemplate;private static String createLockValue() {return String.valueOf(System.currentTimeMillis()) + "@" + RandomStringUtils.randomNumeric(6);}/*** 获取锁** @param key* @param timeOutMillis* @param tryCount* @param tryIntervalMillis* @return boolean*/public boolean tryLock(String key, long timeOutMillis, int tryCount, long tryIntervalMillis) {String lockKey = LOCK_KEY_PREFIX + key;//如果已经获取过锁，则不需再获取if (threadLocal.get() != null) {//计算锁的深度threadLocal.get().incrementAndGet();return true;}for (int i = 0; i < tryCount; i++) {if (redisTemplate.opsForValue().setIfAbsent(lockKey, createLockValue())) {redisTemplate.expire(lockKey, timeOutMillis, TimeUnit.MILLISECONDS);return true;}if (tryCount > 1) {try {Thread.sleep(tryIntervalMillis);} catch (InterruptedException e) {}}}// 防止死锁处理String lockValue = (String) redisTemplate.opsForValue().get(lockKey);if (lockValue != null) {long lockTime = Long.parseLong(lockValue.substring(0, lockValue.indexOf("@")));if (System.currentTimeMillis() - lockTime > timeOutMillis) {//如果并发设置的时候，判断哪个是锁的真正获得者String oldValue = (String) redisTemplate.opsForValue().getAndSet(lockKey, createLockValue());if (lockValue.equals(oldValue)) {redisTemplate.expire(lockKey, timeOutMillis, TimeUnit.MILLISECONDS);return true;}}}return false;}/*** 释放锁** @param key*/public void releaseLock(String key) {redisTemplate.delete(LOCK_KEY_PREFIX + key);threadLocal.remove();}}

保证setnx和expire的复合操作的原子性

因为我们我使用的Redis客户端是Spring Data Redis | RedisTemplate ,同时因为Spring的RedisTemplate的API中不支持Redis的原生命令set多参数版本(Jedis貌似有API支持的)；所以我才需要使用上面的方案；如果你使用了支持set多参数命令行的API，可以保证了setnx和expire两个动作的原子性，你就不需要上面的设计方案了，而是可以采用一种更加简单的设计
set key value [EX seconds] [PX milliseconds] [NX|XX]

说白了，就是可以省去CAS判断锁是否实际过期的情况，因为保证了setnx和expire两个动作的原子性，只要生成锁，那么这把锁就肯定就有过期。因为就可以省下很多的操作，这是一个种比较简单舒服的方式，不需要考虑这么多，同时锁的value也不需要是时间戳+过期时间了，可以是任意不重复的随机数即可

@Component
public class RemoteLock {private final static String KEY = "Redis:Lock:";@AutowiredJedisPool jedisPool;/*** 加锁** @param key* @param ttl* @return*/public boolean tryLock(String key, Long ttl) {Jedis jedis = jedisPool.getResource();String val = System.currentTimeMillis() + "";String result = jedis.set(KEY + key, val, "NX", "EX", ttl);return "OK".equals(result);}/*** 释放锁** @param key* @return*/public boolean releaseLock(String key) {Jedis jedis = jedisPool.getResource();return jedis.del(KEY + key) > 0;}
}

当然，有时候我们的api不支持一些原子组合命令操作的时候，也可以通过Redis的事务或者Lua脚本来封装多个原子操作，因为一个Lua脚本在Redis中就是一个完整的整体，也是具有原子性的。

Redis分布式锁的使用场景分析

参考资料

Distributed locks with Redis - @作者：redis官方
分布式锁解决并发的三种实现方式 - @作者：寂寞的肥皂
Redis setNX 实现分布式锁（重复数据插入可用其来实现排他锁）- @作者：前度刘郎
分布式锁的几种实现方式 - @作者：爷的眼睛闪亮
怎样做可靠的分布式锁，Redlock 真的可行么？ - @作者：YoungChen__