目录
- 前言
- 课程内容
- 一、一个案例引发的思考
- 二、Redis分布式锁的演进
- 2.1 单纯使用Redis的setnx实现分布式锁
- 2.2 setnx + 过期时间
- 3.3 Redisson实现分布式锁：setnx + 过期时间 + 锁续命
- 三、Redisson客户端实现的分布式锁及源码分析
- 学习总结
前言
Redis中间件，非常推荐大家学的一个东西。甚至这么说，Redis也许是我们Java程序员，能接触到的分布式、微服务中间件中一个较为高级，但又比较接地气的中间件了。为什么接地气？因为哪怕是在小项目中，Redis都是一个比较常用、可靠的中间件！
但是我发现，新手用Redis缓存很容易钻入一个牛角尖，那就是Redis会不会崩啊？万一哪一天断电了，宕机了呢？最后得到一个结论：Redis不可靠啊！！！ 但现实是，博主当前所在的小公司小项目Redis生产环境运行2年没蹦过。而且我们那点小体量，就算是崩了也无所谓，重启就行了（事实上，大公司大项目都会使用Redis集群解决这个问题）。但这话说的不严谨，其实关于Redis不可靠问题，是我在前一篇文章说的正是【Redis主从架构】、【Redis集群】要解决的问题，人家已经给出方案了。一句话：切勿讳疾忌医啊同学们。
课程内容
一、一个案例引发的思考
先假设，我们当前线上有一个项目，使用nginx分别轮循到2个tomcat上。它的模型如下：
如上图，为了减缓节点压力，我们把项目部署成了2个tomcat，分别是8080端口和8081端口。并且采用的是轮询策略，客户端每次过来一条请求，将按序依次分流到这2个tomcat上。
然后，这个tomcat项目提供了如下这个接口：
@RequestMapping("/deduct_stock")
public String deductStock1() {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
return "end";
}
案例分析：这个案例很简单，就是提供一个扣减库存的接口，模拟外部电商系统购买物品之后，扣减库存。但是，大家一定要注意到以下几个点：
- 无论我们部署多少个tomcat，库存肯定是被共享的。假设，我只有100个优惠产品，那肯定只能被卖出去100件
- 我i们把库存量放到了redis中去了，每卖出去一件，
stock - 1，并且写回缓存
但事实上，上面这个代码是有问题的。不知道大家能不能理解到？这个对于有经验的朋友来说可能洒洒水而已，很简单。为了照顾萌新，我这边画个图吧。
上图是单个请求（线程）扣减库存的时序图（可能画的不标准，别在意，意思到了就好）。单个线程之下，如果请求都是串行的，也就是上一条执行完了，下一条继续进来请求扣减库存，那当然没问题。但是同学们啊，我们这里是多线程、多个tomcat的分布式环境，所以，你很可能会遇到下面这种情况：
我想，我上图已经画的很清楚了。当这里有另一个客户端请求进来的时候，并且请求顺序跟上面一样，情况显然就开始不对了，出现了【超卖】问题（两个客户端都扣减了一次库存，但是写回都是：99）。为了提点一下小白，上述的并发编程思想，我尽量再点一下：
- 一个http请求，代表着一个线程，所以，同一时刻不同的http请求，肯定是两条不同的线程
- 由于线程之间的隔离，每个线程都会保存一个变量副本，即：上图中客户端1跟客户端2，都会各自保存一个stock副本，各自进行加减操作，然后再把结果写回redis
好了，既然【超卖】问题已经出现了，那上面的问题怎么解决呢？下面，我们就好好研究一下，这个解决方式的演进。
二、Redis分布式锁的演进
一个很正常的思路，对于这种资源共享问题，多线程竞争问题，我想很多同学会想：那就加个锁呗。于是，有朋友提出了：synchronized锁住代码块，嘿嘿，如下所示：
@RequestMapping("/deduct_stock")
public String deductStock1() {
synchronized (this) {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
}
return "end";
}
咱先别讨论【锁粒度】问题，是不是真的有朋友想着用上面这种方式解决的呢？
这么说，上面这种方式在一个tomcat下，单进程的时候，是有效的，但是大伙忘了我们当前的环境，2个tomcat，分布式环境啊！你在tomcat1加synchronized锁，我tomcat2是感知不到的！所以得换个方式。
有经验的朋友可能已经想到了，利用redis io跟命令处理是单线程的特性，所以可以使用setnx key value实现分布式锁，没错，这就是我们现在要讲的东西。下面我们将开始演进，利用Redis实现分布式锁需要解决的问题。这边用的Redis客户端（工具）是StringRedisTemplate，具体使用方式这边就不介绍了。
2.1 单纯使用Redis的setnx实现分布式锁
改正后的代码如下：
@RequestMapping("/deduct_stock")
public String deductStock1() {
String lockKey = "lock:product_101";
// 使用redis的setnx命令
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "zhuge");
if (!result) {
return "error_code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
} finally {
stringRedisTemplate.delete(lockKey);
}
return "end";
}
这边很简单，就是使用了stringRedisTemplate.opsForValue().setIfAbsent，即：Redissetnx命令。然后，如果Redis返回的result不是true，那就返回一个错误码，提示客户端【上锁失败】就好了。那同学们，这样就行了吗？我们画个图吧，嘿嘿嘿
就像上图这样，显然，从目前来看，是没问题，确实已经实现了，多个tomcat情况下，都能控制共享资源了。但是，万一，真的出现了，客户端1在拿到锁之后，还没走到释放锁的时候宕机了，那完了，资源没办法被释放！怎么办？难道我手动删除不成？这就是，单纯利用setnx会遇到的第一个问题：死锁。
哈，我想到了这里，敏锐的同学发现了，既然有这种问题，那我个过期时间不就行了吗？对啊，给过期时间啊，行得通！
2.2 setnx + 过期时间
对于这个方案，其实有2个版本，我先说第一个错误版本：
@RequestMapping("/deduct_stock")
public String deductStock1() {
String lockKey = "lock:product_101";
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "zhuge");
stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
if (!result) {
return "error_code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
} finally {
stringRedisTemplate.delete(lockKey);
}
return "end";
}
跟上述代码一样，我们在setIfAbsent之后，加一个过期时间函数expire。这个方案其实是不行的。很显然，目前这两步操作不是【原子性】的，Java代码嘛，肯定是一条一条按顺序执行的，就跟上面的例子一样，当我们出现极端情况，诶，还真就执行完setIfAbsent之后，expire之前宕机了呢？一样完犊子，会出现死锁，所以，正常我们是利用setIfAbsent另一个重载方法，它会帮我们【原子】地操作这两步，如下：
@RequestMapping("/deduct_stock")
public String deductStock1() {
String lockKey = "lock:product_101";
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "zhuge", 30, TimeUnit.SECONDS);
if (!result) {
return "error_code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
} finally {
stringRedisTemplate.delete(lockKey);
}
return "end";
}
好了，目前是原子性的了。并且呢，我们给这个锁，+了一个30秒的过期时间。可以了吗？啊，不完全对。大家能想明白吗？
很显然啊，你这个过期时间是固定30秒的，万一我业务30秒内完成不了呢？嘿，你是不是想说什么业务30秒完成不了，哈，真可能出现，比如IO阻塞了什么的。
那有朋友会继续建议：那我设置60秒？120秒？240秒？丢，我设置超长时间，总行了吧？？？咳咳咳，啊这个，有点道理的
可如果，我拿出：我不管，你锁多久我业务执行时间永远比你多1秒，阁下该如何应对呢？
哈哈，你是不是想说我无理取闹。好吧，我不抬杠了，我提出一个比较合理的问题哦。
如果你设置的过期时间比较长假设5分钟，万一这时候真的宕机了呢？等过期时间到期吗？天啊，如果此时有数百万个请求进来，你是不是想让人等你5分钟自动解锁啊？这个时候跟宕机有啥区别！
上述说问题的就是由于过期时间比较长，造成的整体【拒绝服务】问题，这是问题一。
其实时间设置过短，我想大家也能想得到会出现什么问题，可能业务没执行完就释放锁了，最后锁形同虚设，其他请求一样进来了，到时候又出现了跟最开始说的情景。这是问题二。
另外还想补充一点，时间设置过短其实还会出现一个很有意思的现象，这里我们画个图给大家看看：
由于客户端A在执行业务期间锁就过期了，此时，客户端B进来加锁肯定是能成功的。但是客户端A在没有出现错误的情况，肯定会继续执行下去的，并且最终会释放锁。那最终这个释放锁释放的是谁的锁呢？客户端B的呀！此时，又有一个新的客户端C过来加锁，那不是成功了吗？显然这样做是有问题的，【错误释放别人的锁】，并且自己的业务还不一定执行完了！
其实针对这个问题，还是有解决方案的。那就是每次上锁的时候，+一个uuid，最后释放锁的时候判断一下uuid是不是跟当前的uuid一样就好了。如下：
@RequestMapping("/deduct_stock")
public String deductStock1() {
String lockKey = "lock:product_101";
String clientId = UUID.randomUUID().toString();
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
if (!result) {
return "error_code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
} finally {
if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
stringRedisTemplate.delete(lockKey);
}
}
return "end";
}
关键代码如上：最后finnaly块判断释放的时候，里面的value值是不是当初我们设置的那个。但其实这仅仅只是解决了我们其中一个问题而已。还有个关键问题，那就是锁时间到底该怎么来确定？
于是有人提出了一个方案：锁续命。顾名思义，就是临到期前，重新设置过期时间。
3.3 Redisson实现分布式锁：setnx + 过期时间 + 锁续命
这里就要开始介绍，基于Redisson实现的一个分布式锁方案了。首先，要使用Reddison，需要先引入jar包，pom.xml如下：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>
代码示例：
@RequestMapping("/deduct_stock")
public String deductStock1() {
String lockKey = "lock:product_001";
//获取锁对象
RLock redissonLock = redisson.getLock(lockKey);
//加分布式锁
redissonLock.lock(); // .setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功，剩余库存:" + realStock);
} else {
System.out.println("扣减失败，库存不足");
}
} finally {
//解锁
redissonLock.unlock();
}
return "end";
}
上面的代码很简单，我们无需关心之前提到的那几个问题了，Redisson在封装的api里面已经帮我们做好了一切。我们只需要简单的调用lock跟unlock而已。
三、Redisson客户端实现的分布式锁及源码分析
待完成。。。
学习总结
- 学习了Redis分布式锁实现的演进过程
- 学习了基于Redis实现分布式锁需要解决的一些问题，及解决思路
