Redis 分布式锁

概述

单机架构下，一个进程中的多个线程竞争同一共享资源时，通常使用 JVM 级别的锁即可保证互斥，以对商品下单并扣库存为例：

1. public String deductStock() {
2.     synchronized (this){
3.         // 获取库存值
4.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
5.         if (stock > 0) {
6.             int realStock = stock - 1;
7.             stringRedisTemplate.opsForValue().set("stock", realStock + "")
8.             System.out.println("扣减成功，剩余库存:" + realStock);
9.         } else {
10.             System.out.println("扣减失败，库存不足");
11.         }
12.     }
13.     return "end";
14. }

复制代码

然而，当使用分布式架构时，这种方式就不管用了，因为 JVM 锁只能控制自家应用，其他机器的应用时管不了的，这时候分布式锁就派上用场了，它能保证分布式系统下不同进程对共享资源访问的互斥性

案例分析

下面对使用 Redis 实现分布式锁的案例进行分析：
1. Case1

使用 Redis 中的 setnx() 设计一个入门级别的分布式锁

1. public String deductStock1() {
2.     String localKey = "lock:product:0001";
3.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, "true");
4.     if (!aBoolean){
5.         return "当前系统繁忙";
6.     }
7.     try {
8.         // 获取库存值
9.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
10.         if (stock > 0) {
11.             int realStock = stock - 1;
12.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
13.             System.out.println("扣减成功，剩余库存:" + realStock);
14.         } else {
15.             System.out.println("扣减失败，库存不足");
16.         }
17.     } finally {
18.         // 即使中间的任何一处逻辑抛出异常,也能保证锁释放
19.         stringRedisTemplate.delete(localKey);
20.     }
21.     return "end";
22. }

复制代码

存在的问题：锁没有释放，机器却宕机了，这时候其他机器将无法获取锁
2. Case2

设置一个过期时间，解决 Case1 中存在的宕机没有释放锁的问题

1. public String deductStock2() {
2.     String localKey = "lock:product:0001";
3.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, "true");
4.     stringRedisTemplate.expire(localKey,10,TimeUnit.SECONDS);
5.     if (!aBoolean){
6.         return "当前系统繁忙";
7.     }
8.     try {
9.         // 获取库存值
10.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
11.         if (stock > 0) {
12.             int realStock = stock - 1;
13.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
14.             System.out.println("扣减成功，剩余库存:" + realStock);
15.         } else {
16.             System.out.println("扣减失败，库存不足");
17.         }
18.     } finally {
19.         stringRedisTemplate.delete(localKey);
20.     }
21.     return "end";
22. }

复制代码

存在的问题：有可能还没有执行到 expire() 就宕机了，没有保证原子性
3. Case3

在加锁时就设置超时时间，保证加锁和设置超时时间是原子操作

1. public String deductStock3() {
2.     String localKey = "lock:product:0001";
3.     // 这条命令能够保证原子性
4.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, "true",10,TimeUnit.SECONDS);
5.     if (!aBoolean){
6.         return "当前系统繁忙";
7.     }
8.     try {
9.         // 获取库存值
10.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
11.         if (stock > 0) {
12.             int realStock = stock - 1;
13.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
14.             System.out.println("扣减成功，剩余库存:" + realStock);
15.         } else {
16.             System.out.println("扣减失败，库存不足");
17.         }
18.     } finally {
19.         stringRedisTemplate.delete(localKey);
20.     }
21.     return "end";
22. }

复制代码

存在问题：如果系统并发量不是特别的大，那么问题不大，但如果并发量很大，就会出现严重的并发问题：

• 假设线程 A 的时间超过了超时时间，锁失效了，此时该线程 A 还没有执行 delete 方法
  
• 线程 B 这时候加锁成功了，与此同时线程 A 执行了 delete 方法，但是这时候线程 A 释放的锁是线程 B 的
  
• 于是极端情况下就会出现：线程 A 释放线程 B 的锁，B 释放 C 的，C 释放 D 的 ......
  

4. Case4

Case3 存在的问题的根本原因就是在执行 delete 方法的时候，自己的锁被其他的线程释放了，所以解决办法就是给每个线程生成一个唯一 ID，在最后释放锁的时候判断是否是自己的锁，如果是自己的才释放

1. public String deductStock4() {
2.     String localKey = "lock:product:0001";
3.     String uuid = UUID.randomUUID().toString();
4.     // 这条命令能够保证原子性
5.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, uuid,10,TimeUnit.SECONDS);
6.     if (!aBoolean){
7.         return "当前系统繁忙";
8.     }
9.     try {
10.         // 获取库存值
11.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
12.         if (stock > 0) {
13.             int realStock = stock - 1;
14.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
15.             System.out.println("扣减成功，剩余库存:" + realStock);
16.         } else {
17.             System.out.println("扣减失败，库存不足");
18.         }
19.     } finally {
20.         if (uuid.equals(stringRedisTemplate.opsForValue().get(localKey))){
21.             stringRedisTemplate.delete(localKey);
22.         }
23.     }
24.     return "end";
25. }

复制代码

存在问题：存在原子性问题，问题代码如下：

1. if (uuid.equals(stringRedisTemplate.opsForValue().get(localKey))){
2.     stringRedisTemplate.delete(localKey);
3. }

复制代码

有可能出现当前线程执行完 if 判断却还没执行 delete 操作的时候当前锁过期了，于是又会出现当前线程释放了其他线程的锁的情况
5. Case5

对于 Case4 的问题，本质是 「判断是不是当前线程加的锁」和「释放锁」不是一个原子操作，可以用 Lua 脚本代替，Redis 会将整个脚本作为一个整体执行

1. String redisScript = "
2.     if redis.call('get',KEYS[1]) == ARGV[1] then
3.         return redis.call('del',KEYS[1])
4.     else
5.         return 0
6.         end;"
8. public String deductStock5() {
9.     String localKey = "lock:product:0001";
10.     String uuid = UUID.randomUUID().toString();
11.     // 这条命令能够保证原子性
12.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, uuid,10,TimeUnit.SECONDS);
13.     if (!aBoolean){
14.         return "当前系统繁忙";
15.     }
16.     try {
17.         // 获取库存值
18.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
19.         if (stock > 0) {
20.             int realStock = stock - 1;
21.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
22.             System.out.println("扣减成功，剩余库存:" + realStock);
23.         } else {
24.             System.out.println("扣减失败，库存不足");
25.         }
26.     } finally {
27.         redisTemplate.execute(redisScript, Arrays.asList(localKey), uuid);
28.     }
29.     return "end";
30. }

复制代码

也可以使用锁续命的方式解决，即创建一个守护线程，每过一段时间，判断业务的主线程有没有结束（是否还加着锁），如果还加着锁，将锁的超时时间重新设置

1. public String deductStock5() {
2.     String localKey = "lock:product:0001";
3.     String uuid = UUID.randomUUID().toString();
4.     // 这条命令能够保证原子性
5.     Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(localKey, uuid,10,TimeUnit.SECONDS);
6.     if (!aBoolean){
7.         return "当前系统繁忙";
8.     } else {
9.         // 续命
10.         Thread demo = new Thread(new Runnable() {
11.             @Override
12.             public void run() {
13.                 while (true) {
14.                     Boolean expire = redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
15.                     // 有可能已经主动删除key,不需要在续命
16.                     if(!expire){
17.                         return;
18.                     }
19.                     try {
20.                         Thread.sleep(1000);
21.                     } catch (InterruptedException e) {
22.                         e.printStackTrace();
23.                     }
24.                 }
25.             }
26.         });
27.         demo.setDaemon(true);
28.         demo.start();
29.     }
30.     try {
31.         // 获取库存值
32.         int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
33.         if (stock > 0) {
34.             int realStock = stock - 1;
35.             stringRedisTemplate.opsForValue().set("stock", realStock + "");
36.             System.out.println("扣减成功，剩余库存:" + realStock);
37.         } else {
38.             System.out.println("扣减失败，库存不足");
39.         }
40.     } finally {
41.         if (uuid.equals(stringRedisTemplate.opsForValue().get(localKey))){
42.             stringRedisTemplate.delete(localKey);
43.         }
44.     }
45.     return "end";
46. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！