Redis高并发分布式锁详解

🚀 优质资源分享 🚀

为什么需要分布式锁

1.为了解决Java共享内存模型带来的线程安全问题，我们可以通过加锁来保证资源访问的单一，如JVM内置锁synchronized，类级别的锁ReentrantLock。

2.但是随着业务的发展，单机服务毕竟存在着限制，故会往多台组合形成集群架构，面对集群架构，我们同样存在则资源共享问题，而每台服务器有着自己的JVM，这时候我们对于锁的实现不得不考虑分布式的实现。

分布式锁应该具备哪些条件

1.在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行

2.高可用的获取锁与释放锁

3.高性能的获取锁与释放锁

4.具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）

5.具备锁失效机制，即自动解锁，防止死锁

6.具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

秒杀抢购场景模拟（模拟并发问题：其实就是指每一步如果存在间隔时间，那么当某一线程间隔时间拉长，会对其余线程造成什么影响）

0.如果要在本机测试的话

1）配置Nginx实现负载均衡

http {
 upstream testfuzai {
 server 127.0.0.1:8080 weight=1;
 server 127.0.0.1:8090 weight=1;
 }
 
 server {
 listen 80;
 server\_name localhost;
 
 location / {
 //proxy\_pass：设置后端代理服务器的地址。这个地址(address)可以是一个域名或ip地址和端口，或者一个 unix-domain socket路径。
            proxy\_pass http://testfuzai;
 proxy\_set\_header Host $proxy\_host;
 }
 }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2）启动redis设置好参数与数量

3）启动项目并分别配置不同端口（要与Nginx里面的一致）

4）进行压测，通过jmeter的Thread Group里面编辑好HTTP Request，设置参数 线程数 Number of Threads 【设置为200】 ，请求的重复次数 Loop count 【设置为5】 ，Ramp-up period（seconds）线程启动开始运行的时间间隔（单位是秒）【设置为1】。则，一秒内会有1000个请求打过去。

1.不加锁进行库存扣减的情况：

代码示例

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

发现说明

1）通过打印输出，我们会发现两台机器上会出现重复的值（即出现了超卖现象）。甚至会出现另一台服务器的数据覆盖本服务器的数据。

2）原因在于读取数据和写入数据存在时间差，如两个服务器Q1和Q1，Q1有请求，获取库存【假设300】，在库存判断大小之后进行扣减库存如果慢了【假设需要3秒】，那么Q2有5次请求，获取到库存，扣减完后设置，依次5次，则库存为【295】。但是此时Q1完成自身请求又会把库存设置为【299】。故不合理。所以应该改为使用stringRedisTemplate.boundValueOps(“stock”).increment(-1); 改为采用redis内部扣除，减少了超卖的个数。但是就算改了也只是避免了覆盖问题，仍然没有解决超卖问题。如果有6台服务器，库存剩下1个的时候六个请求同时进入到扣减库存这一步，那么就会出现超卖5个的现象（这也是超卖个数最多的现象）。

2.采用SETNX的方式加分布式锁的情况：

代码示例

public String deductStock() {
 String lockKey = "lock:product\_101";
 Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockKey);if (!result) {
 return "error\_code";
 }

 try {　　　　 int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); 
        if (stock > 0) {
 Long realStock = (Long) stringRedisTemplate.opsForValue().decrement("stock"); 
            System.out.println("扣减成功，剩余库存:" + realStock);
 } else {
 System.out.println("扣减失败，库存不足");
 }
 } finally {
 stringRedisTemplate.delete(lockKey);
 }
 return "end";
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

发现说明

1）这种方式明显保证了在分布式情况下只有一个线程能够执行业务代码。但是我们不可能对于用户买商品的时候返回错误提示，如果不断自旋的话又容易让CPU飙升。肯定要考虑休眠与唤醒，但可以在上层方法里面处理。

2）同时很明显存在个问题，如果我在扣减库存时候服务器宕机了，库存扣减还没设置【且没执行finally代码，那么我这个商品的锁就不会被释放，除非手动清除】。

那么肯定需要设置超时时间。如

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, lockKey);
stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);
1
2

会发现补一个超时时间的话依旧无法避免之前的问题，故加锁和设置超时时间需要保持原子性。

3）采用原子操作：Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);基于设置了超时时间，那么我们如何考量超时时间呢，业务执行多久我们根本不可得知。故容易出现时间到期了，业务还没执行完。这就容易出现A持有锁执行任务，还没完成就超时了，B持有锁执行任务，A执行完，释放锁【此时会释放B的锁】的情况。所以释放锁必须要持有锁本人才能执行。

if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
 stringRedisTemplate.delete(lockKey);
}
1
2
3

所以clientId需要是分布式ID，然后释放锁改为判断clientId符合才能去释放。

3**.改进之后的情况：**

代码示例

public String deductStock() {
 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 {
1
2
3
4
5
6
7
8
9

　　　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); 
1

 if (stock > 0) { Long realStock = (Long) stringRedisTemplate.opsForValue().decrement("stock");  System.out.println("扣减成功，剩余库存:" + realStock); } else { System.out.println("扣减失败，库存不足"); } } finally { if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) { stringRedisTemplate.delete(lockKey); } } return "end"; }
1

发现说明

1）即时加了判断，我们会发现依旧会存在问题【因为判断与释放锁操作不是原子性的】，如果在判断里面加上休眠进行试验

if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
 Thread.sleep(20000);
 stringRedisTemplate.delete(lockKey);
}
1
2
3
4

我们会发现根本问题依旧没有解决，只是减少了发生的情况。究其原因