redis实战---分布式锁--实战篇

女爷i 2023-10-09 16:12 38阅读 0赞

#### 分布式锁实战 ####

*  故事背景
 *  问题复现
 *  解决方案
 *   *  自己手动实现
     *   *  代码
         *  压测结果
         *  代码重点解释
     *  基于Redisson进行实现
     *   *  引入依赖
         *  代码使用
         *  运行结果
         *  源码解析
 *  总结&升华

## 故事背景 ##

[上文][Link 1]讲到我们使用**synchronized**实现了jvm级别的加锁。同时抛出了在分布式环境下，我们的代码会出现的问题。这篇文章，将会带着大家去解决这个问题。带着大家一起实现redis的分布式锁。  
![在这里插入图片描述][6b4ac8b1ffaf440daa0c30caf01a7b27.png]

## 问题复现 ##

1.  官网上下载nginx
2.  配置负载均衡。  
    ![在这里插入图片描述][7768d44baab948288adefebd6332cb81.png]

# 定义 upstream 池
        upstream backend {
            server 127.0.0.1:8000;
            server 127.0.0.1:8001;
        }
         	location / {
            proxy_pass http://backend;
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    		}

我将我的项目复制了一分，修改了端口号，模拟部署在不同的环境上。以上配置将会以轮训的方式进行服务调用  
3. 接口调用接口  
![在这里插入图片描述][5bcfc4c2891b4f8db1bcefcf5ac77223.png]  
可以看到，两个服务都分别被调用了一次。这时候我们使用的**synchronized**锁就会失效了，下面我们压测一下，看看结果。  
4. 压测，复现问题。  
![在这里插入图片描述][9ec734b072b24e2f85c6591ed794da17.png]  
结果：  
![在这里插入图片描述][51bf7235321049d4be06762cbbdcdc48.png]  
上图中，左边是端口8001的服务，右边是8000的服务，我们发现，这两个服务虽然单独的看，销售的商品都是正确的，但是放在一起看，就会发现有相同的库存，这就说明，**同一个库存被卖了两次**，我们上文提到的超卖问题仍然存在！！

## 解决方案 ##

### 自己手动实现 ###

下面将会讲解如何自己进行手动实现分布式锁，此方式只供大家参考理解，如果项目上用，推荐集成Redisson使用其提供的解决方案。

#### 代码 ####

@RestController
    @RequestMapping("/test")
    public class IndexController {
        
    
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
    
        @GetMapping("lock")
        public  String deductStock() {
        
    
            //写死一个固定的商品id，作为我们被秒杀的商品
            String lockKey="lock:product:101";
            //uuid,防止删除其他人加的锁
            String clientId = UUID.randomUUID().toString();
            //进行加锁，设置过期时间为10s 注意代码的原子性
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,clientId,10, TimeUnit.SECONDS);
            //如果加锁失败,返回错误，秒未成功
            if(!result){
        
                return "error_code";
            }
    
            try {
        
    
                // 获取当前库存
                String stock1 = stringRedisTemplate.opsForValue().get("stock");
                if( stock1 == null){
        
                    System.out.println("秒杀未开始");
                    return "end";
                }
    
                int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
                if (stock > 0) {
        
                    // 扣减库存
                    int realStock = stock - 1;
                    // 更新库存
                    stringRedisTemplate.opsForValue().set("stock", realStock + "");
                    System.out.println("扣减成功，剩余的库存为：" + realStock);
    
                } else {
        
                    System.out.println("扣减失败，库存不足");
                }
    
            }finally {
        
                //如果是自己加的锁就自己删掉
                if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
        
                    stringRedisTemplate.delete(lockKey);
                }
            }
    
            return "end";
        }
    
    }

#### 压测结果 ####

![在这里插入图片描述][52b89f342e7f4de78c7c02ec466eb196.png]  
两个项目，没有超卖问题。一共20个请求，其中6个加锁成功，消耗了库存，其余14个由于没有加锁成功，秒杀失败。

#### 代码重点解释 ####

1.使用setnx  
SETNX 是 Redis 数据库中的一个命令，用于将一个键值对（key-value pair）设置到 Redis 中，但只有在该键不存在的情况下才会设置成功。如果该键已经存在，SETNX 命令不会对其进行任何操作，并返回 0，否则返回 1。  
2.死锁问题  
使用setnx进行加锁的时候，一定要设置锁的过期时间。业务完成之后，一定要及时释放锁，避免产生死锁问题。并且一定要保证加锁和设置锁的过期时间操作是原子的，避免只上锁，未设置过期时间问题的存在  
3.锁续命问题  
上述代码作为一个简单的分布式锁实现，在并发量不算很高的情况下，不会出现什么问题，但是它实际上还是有瑕疵的。我们上述代码，锁失效有两种可能。一种是过期，另一种是代码删除。代码删除没什么问题，我们选择将锁删除的时候，肯定是业务代码执行完毕。但是如果是过期的话，**有可能我们的业务代码还没有执行完，锁先过期了**，并发量大的情况下，外部不断有请求试图加锁，可能会造成锁失效的情况。

### 基于Redisson进行实现 ###

我们可以通过为锁续命来解决上文所述，**代码未执行完毕，锁已经过期**的问题，这里将使用Redisson的解决方案

#### 引入依赖 ####

<dependency>
                <groupId>org.redisson</groupId>
                <artifactId>redisson-spring-boot-starter</artifactId>
                <version>3.15.5</version>
            </dependency>

#### 代码使用 ####

@Autowired
        private RedissonClient redissonClient;
    
        @GetMapping("lock1")
        public  String deductStock1() {
        
    
            //写死一个固定的商品id，作为我们被秒杀的商品
            String lockKey="lock:product:101";
            //获取锁对象
            RLock lock = redissonClient.getLock(lockKey);
            //加锁，使用lock方法，锁将会自动续命
            lock.lock();
            try {
        
                // 获取当前库存
                String stock1 = stringRedisTemplate.opsForValue().get("stock");
                if( stock1 == null){
        
                    System.out.println("秒杀未开始");
                    return "end";
                }
    
                int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
                if (stock > 0) {
        
                    // 扣减库存
                    int realStock = stock - 1;
                    // 更新库存
                    stringRedisTemplate.opsForValue().set("stock", realStock + "");
                    System.out.println("扣减成功，剩余的库存为：" + realStock);
    
                } else {
        
                    System.out.println("扣减失败，库存不足");
                }
                return "end";
            }finally {
        
                //释放锁
                lock.unlock();
            }
        }

#### 运行结果 ####

我们来压测一下，发现可以实现分布式锁的效果，不会出现超卖问题  
![在这里插入图片描述][db49e5543d994004806a123752f11170.png]

#### 源码解析 ####

Redisson的实现锁的续命，主要的代码在 RedissonLock类的lock方法中，下面我们来解析下它的lock方法

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        
        // 获取当前线程ID
        long threadId = Thread.currentThread().getId();
        // 尝试获取锁，获取到了则返回null，否则返回锁的剩余过期时间
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // 如果返回null，说明锁已经被当前线程获取，直接返回
        if (ttl == null) {
        
            return;
        }
    
        // 创建一个订阅对象
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        // 如果可中断，则中断等待
        if (interruptibly) {
        
            commandExecutor.syncSubscriptionInterrupted(future);
        } else {
        
            // 否则一直等待
            commandExecutor.syncSubscription(future);
        }
    
        try {
        
            while (true) {
        
                // 再次尝试获取锁，获取到了则返回null，否则返回锁的剩余过期时间
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // 如果返回null，说明锁已经被当前线程获取，跳出循环
                if (ttl == null) {
        
                    break;
                }
    
                // 如果锁剩余过期时间大于等于0，则等待指定时间
                if (ttl >= 0) {
        
                    try {
        
                        // 等待指定时间后再次尝试获取锁
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
        
                        // 如果是可中断的，则抛出异常
                        if (interruptibly) {
        
                            throw e;
                        }
                        // 否则继续等待
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
        
                    // 否则一直等待直到获取到锁
                    if (interruptibly) {
        
                        future.getNow().getLatch().acquire();
                    } else {
        
                        future.getNow().getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
        
            // 取消订阅
            unsubscribe(future, threadId);
        }
    }

在这段代码中，首先获取了当前线程的ID，并通过 tryAcquire() 方法尝试获取锁，如果获取成功（即 ttl == null），则直接返回。

如果获取不到锁，会进行如下操作：

1.  通过 subscribe() 方法，向 Redisson 客户端发送一条消息，表示当前线程正在等待该锁的释放。
2.  如果 interruptibly 为 true，则使用 syncSubscriptionInterrupted() 方法等待消息；否则，使用 syncSubscription() 方法等待消息。
3.  进入循环，不断尝试获取锁（即调用 tryAcquire() 方法）。
4.  如果获取到锁，则根据 ttl 值进行等待：  
    如果 ttl 大于等于 0，则等待 ttl 时间，同时等待 Redisson 客户端发送消息，通知当前线程释放锁。  
    如果 ttl 小于 0，则说明当前线程已经在 Redisson 客户端的等待队列中，直接等待通知即可。
5.  当获取到锁时，会解除之前发送的等待消息，然后退出循环。

在上述过程中，由于不断地尝试获取锁，因此每次成功获取锁时都会重置锁的过期时间。这样就可以实现锁的自动续命了。

## 总结&升华 ##

通过本篇文章，我们了解到了如何实现redis的分布式锁。学习了如何使用Redisson进行分布式锁，并且解决了锁的续命问题。  
[redis实战][redis]

[Link 1]: https://blog.csdn.net/hlzdbk/article/details/129892338
[6b4ac8b1ffaf440daa0c30caf01a7b27.png]: https://img-blog.csdnimg.cn/6b4ac8b1ffaf440daa0c30caf01a7b27.png
[7768d44baab948288adefebd6332cb81.png]: https://img-blog.csdnimg.cn/7768d44baab948288adefebd6332cb81.png
[5bcfc4c2891b4f8db1bcefcf5ac77223.png]: https://img-blog.csdnimg.cn/5bcfc4c2891b4f8db1bcefcf5ac77223.png
[9ec734b072b24e2f85c6591ed794da17.png]: https://img-blog.csdnimg.cn/9ec734b072b24e2f85c6591ed794da17.png
[51bf7235321049d4be06762cbbdcdc48.png]: https://img-blog.csdnimg.cn/51bf7235321049d4be06762cbbdcdc48.png
[52b89f342e7f4de78c7c02ec466eb196.png]: https://img-blog.csdnimg.cn/52b89f342e7f4de78c7c02ec466eb196.png
[db49e5543d994004806a123752f11170.png]: https://img-blog.csdnimg.cn/db49e5543d994004806a123752f11170.png
[redis]: https://blog.csdn.net/hlzdbk/category_12270164.html?spm=1001.2014.3001.5482