Redis分布式锁实战
啥是分布式锁？
我们学习 Java 都知道锁的概念，例如基于 JVM 实现的同步锁 synchronized，以及 jdk 提供的一套代码级别的锁机制 lock，我们在并发编程中会经常用这两种锁去保证代码在多线程环境下运行的正确性。但是这些锁机制在分布式场景下是不适用的，原因是在分布式业务场景下，我们的代码都是跑在不同的JVM甚至是不同的机器上，synchronized 和 lock 只能在同一个 JVM 环境下起作用。所以这时候就需要用到分布式锁了。
例如，现在有个场景就是整点抢消费券（疫情的原因，支付宝最近在8点、12点整点开放抢消费券），消费券有一个固定的量，先到先得，抢完就没了，线上的服务都是部署多个的，大致架构如下：
所以这个时候我们就得用分布式锁来保证共享资源的访问的正确性。
回到顶部
为什么要用分布式锁嗯？
假设不使用分布式锁，我们看看 synchronized 能不能保证？其实是不能的，我们来演示一下。
下面我写了一个简单的 springboot 项目来模拟这个抢消费券的场景，代码很简单，大致意思是先从 Redis 获取剩余消费券数，然后判断大于0，则减一模拟被某个用户抢到一个，然后减一后再修改 Redis 的剩余消费券数量，打印扣减成功，剩余还有多少，否则扣减失败，就没抢到。整块代码被 synchronized 包裹，Redis 设置的库存数量为50。
//假设库存编号是00001
private String key = "stock:00001";
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 扣减库存 synchronized同步锁
*/
@RequestMapping("/deductStock")
public String deductStock(){
synchronized (this){
//获取当前库存
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
if(stock>0){
int afterStock = stock-1;
stringRedisTemplate.opsForValue().set(key,afterStock+"");//修改库存
System.out.println("扣减库存成功，剩余库存"+afterStock);
}else {
System.out.println("扣减库存失败");
}
}
return "ok";
}
然后启动两个springboot项目，端口分别为8080,8081，然后在nginx里配置负载均衡
upstream redislock{
server 127.0.0.1:8080;
server 127.0.0.1:8081;
}
server {
listen 80;
server_name 127.0.0.1;
location / {
root html;
index index.html index.htm;
proxy_pass http://redislock;
}
}
然后用jmeter压测工具进行测试
然后我们看一下控制台输出，可以看到我们运行的两个web实例，很多同样的消费券被不同的线程抢到，证明synchronized在这样的情况下是不起作用的，所以就需要使用分布式锁来保证资源的正确性。
回到顶部
如何用Redis实现分布式锁？
在实现分布式锁之前，我们先考虑如何实现，以及都要实现锁的哪些功能。
1、分布式特性（部署在多个机器上的实例都能够访问这把锁）
2、排他性（同一时间只能有一个线程持有锁）
3、超时自动释放的特性（持有锁的线程需要给定一定的持有锁的最大时间，防止线程死掉无法释放锁而造成死锁）
4、...
基于以上列出的分布式锁需要拥有的基本特性，我们思考一下使用Redis该如何实现？
1、第一个分布式的特性Redis已经支持，多个实例连同一个Redis即可
2、第二个排他性，也就是要实现一个独占锁，可以使用Redis的setnx命令实现
3、第三个超时自动释放特性，Redis可以针对某个key设置过期时间
4、执行完毕释放分布式锁
科普时间
Redis Setnx 命令
Redis Setnx（SET if Not eXists） 命令在指定的 key 不存在时，为 key 设置指定的值
语法
redis Setnx 命令基本语法如下：
redis 127.0.0.1:6379> SETNX KEY_NAME VALUE
可用版本：>= 1.0.0
返回值：设置成功，返回1， 设置失败，返回0
@RequestMapping("/stock_redis_lock")
public String stock_redis_lock(){
//底层使用setnx命令
Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key, "true");
stringRedisTemplate.expire(lock_key,10, TimeUnit.SECONDS);//设置过期时间10秒
if (!aTrue) {//设置失败则表示没有拿到分布式锁
return "error";//这里可以给用户一个友好的提示
}
//获取当前库存
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
if(stock>0){
int afterStock = stock-1;
stringRedisTemplate.opsForValue().set(key,afterStock+"");
System.out.println("扣减库存成功，剩余库存"+afterStock);
}else {
System.out.println("扣减库存失败");
}
stringRedisTemplate.delete(lock_key);//执行完毕释放分布式锁
return "ok";
}
仍然设置库存数量为50，我们再用jmeter测试一下，把jmeter的测试地址改为127.0.0.1/stock_redis_lock，同样的设置再来测一次。
测试了5次没有出现脏数据，把发送时间改为0，测了5次也没问题，然后又把线程数改为600，时间为0 ，循环4次，测了几次也是正常的。
上面实现分布式锁的代码已经是一个较为成熟的分布式锁的实现了，对大多数软件公司来说都已经满足需求了。但是上面代码还是有优化的空间，例如:
1）上面的代码我们是没有考虑异常情况的，实际情况下代码没有这么简单，可能还会有别的很多复杂的操作，都有可能会出现异常，所以我们释放锁的代码需要放在finally块里来保证即使是代码抛异常了释放锁的代码他依然会被执行。
2）还有，你有没有注意到，上面我们的分布式锁的代码的获取和设置过期时间的代码是两步操作第4行和第5行，即非原子操作，就有可能刚执行了第4行还没来得及执行第5行这台机器挂了，那么这个锁就没有设置超时时间，其他线程就一直无法获取，除非人工干预，所以这是一步优化的地方，Redis也提供了原子操作，那就是SET key value EX seconds NX
科普时间
SET key value [EX seconds] [PX milliseconds] [NX|XX] 将字符串值 value 关联到 key
可选参数
从 Redis 2.6.12 版本开始， SET 命令的行为可以通过一系列参数来修改：
EX second ：设置键的过期时间为 second 秒。SET key value EX second 效果等同于 SETEX key second value
PX millisecond ：设置键的过期时间为 millisecond 毫秒。SET key value PX millisecond 效果等同于 PSETEX key millisecond value
NX ：只在键不存在时，才对键进行设置操作。SET key value NX 效果等同于 SETNX key value
XX ：只在键已经存在时，才对键进行设置操作
SpringBoot的StringRedisTemplate也有对应的方法实现，如下代码：
//假设库存编号是00001
private String key = "stock:00001";
private String lock_key = "lock_key:00001";
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/stock_redis_lock")
public String stock_redis_lock() {
String uuid = UUID.randomUUID().toString();
try {
//原子的设置key及超时时间
Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key, "true", 30, TimeUnit.SECONDS);
if (!aTrue) {
return "error";
}
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));
if (stock > 0) {
int afterStock = stock - 1;
stringRedisTemplate.opsForValue().set(key, afterStock + "");
System.out.println("扣减库存成功，剩余库存" + afterStock);
} else {
System.out.println("扣减库存失败");
}
} catch (NumberFormatException e) {
e.printStackTrace();
} finally {
//避免死锁
if (uuid.equals(stringRedisTemplate.opsForValue().get(lock_key))) {
stringRedisTemplate.delete(lock_key);
}
}
return "ok";
}
这样实现是否就完美了呢？嗯，对于并发量要求不高或者非大并发的场景的话这样实现已经可以了。但是对于抢购 ，秒杀这样的场景，当流量很大，这时候服务器网卡、磁盘IO、CPU负载都可能会达到极限，那么服务器对于一个请求的的响应时间势必变得比正常情况下慢很多，那么假设就刚才设置的锁的超时时间为10秒，如果某一个线程拿到锁之后因为某些原因没能在10秒内执行完毕锁就失效了，这时候其他线程就会抢占到分布式锁去执行业务逻辑，然后之前的线程执行完了，会去执行 finally 里的释放锁的代码就会把正在占有分布式锁的线程的锁给释放掉，实际上刚刚正在占有锁的线程还没执行完，那么其他线程就又有机会获得锁了...这样整个分布式锁就失效了，将会产生意想不到的后果。如下图模拟了这个场景。
所以这个问题总结一下，就是因为锁的过期时间设置的不合适或因为某些原因导致代码执行时间大于锁过期时间而导致并发问题以及锁被别的线程释放，以至于分布式锁混乱。在简单的说就是两个问题，1）自己的锁被别人释放 2）锁超时无法续时间。
第一个问题很好解决，在设置分布式锁时，我们在当前线程中生产一个唯一串将value设置为这个唯一值，然后在finally块里判断当前锁的value和自己设置的一样时再去执行delete，如下：
String uuid = UUID.randomUUID().toString();
try {
//原子的设置key及超时时间，锁唯一值
Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key,uuid,30,TimeUnit.SECONDS);
//...
} finally {
//是自己设置的锁再执行delete
if(uuid.equals(stringRedisTemplate.opsForValue().get(lock_key))){
stringRedisTemplate.delete(lock_key);//避免死锁
}
}
问题一解决了（设想一下上述代码还有什么问题，一会儿讲），那锁的超时时间就很关键了，不能太大也不能太小，这就需要评估业务代码的执行时间，比如设置个10秒，20秒。即使是你的锁设置了合适的超时时间，也避免不了可能会发生上述分析的因为某些原因代码没在正常评估的时间内执行完毕，所以这时候的解决方案就是给锁续超时时间。大致思路就是，业务线程单独起一个分线程，定时去监听业务线程设置的分布式锁是否还存在，存在就说明业务线程还没执行完，那么就延长锁的超时时间，若锁已不存在则业务线程执行完毕，然后就结束自己。
“锁续命”的这套逻辑属实有点复杂啊，要考虑的问题太多了，稍不注意就会有bug。不要看上面实现分布式锁的代码没有几行，就认为实现起来很简单，如果说自己去实现的时候没有实际高并发的经验，肯定也会踩很多坑，例如，
1）锁的设置和过期时间的设置是非原子操作的，就可能会导致死锁。
2）还有上面遗留的一个，在finally块里判断锁是否是自己设置的，是的话再删除锁，这两步操作也不是原子的，假设刚判断完为true服务就挂了，那么删除锁的代码不会执行，就会造成死锁，即使是设置了过期时间，在没过期这段时间也会死锁。所以这里也是一个注意的点，要保证原子操作的话，Redis提供了执行Lua脚本的功能来保证操作的原子性，具体怎么使用不再展开。
————————————————
版权声明：本文为CSDN博主「weixin_43144260」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_43144260/article/details/107334615