深入浅出超详细从线程锁到 redis 实现分布式锁（篇节 2）「建议收藏」

全栈程序员站长

发布于 2022-08-31 16:38:51

4020

发布于 2022-08-31 16:38:51

文章被收录于专栏：全栈程序员必看全栈程序员必看

大家好，又见面了，我是你们的朋友全栈君。

redis 实现分布式锁

上节我们讲了线程锁在单体项目中的作用，和放在分布式项目里产生的问题，那接下来我们就来解决分布式架构上怎么保证数据的一直性

使用 redisTemplate 实现

// 设置锁
setIfAbsent("lock", "1213")---> SETNX lock "1213"
// 释放锁
redisTemplate.delete("lock");

@GetMapping("/cut")
    public Object kc() { 
   
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        /** * 设置锁，如果 lock 不存在的话，设置 lock=1213 并返回 true * 如果存在的话：就不操作 直接返回 false */
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "1213");
        if(!lock){ 
   
            // 锁存在
            return "服务繁忙，请稍后再试";
        }

        int num = Integer.parseInt(redisTemplate.opsForValue().get("num").toString());
        if (num > 0) { 
   
            int lastNum = num - 1;
            redisTemplate.opsForValue().set("num", lastNum + "");
            System.out.println("扣减库存成功，剩余库存为：" + lastNum);
        } else { 
   
            System.out.println("扣减库存失败，库存不足");
        }

        // 释放锁
        redisTemplate.delete("lock");

        return "ok";
    }

简述一下逻辑：第一个请求进来，设置一把锁进行锁住，然后往下执行减库存的操作，此时第二个线程进来获取锁，但是第一请求没有释放锁，所以第二个请求获取锁就会失败得到返回值为 false，进入if 方法体直接返回了。其他线程也是如此，直到第一个线程释放锁后其他线程才有获取锁的机会，每次只有一个线程能够成功获取锁，其他线程获取不到直接返回。

大家觉得这把锁怎么样，是不是解决问题了呢？还有什么问题吗？有没有那个小可爱想到了呢？

肯定有小伙伴想到了。

情况1：一个请求进来，获取锁，执行扣减库存的操作，在它释放锁之前服务突然抛异常了呢？

那就使用 try{}finally{}嘛，这下不怕抛异常无法释放锁了吧

try{ 
   
       int num = Integer.parseInt(redisTemplate.opsForValue().get("num").toString());
       if (num > 0) { 
   
           int lastNum = num - 1;
           redisTemplate.opsForValue().set("num", lastNum + "");
           System.out.println("扣减库存成功，剩余库存为：" + lastNum);
       } else { 
   
           System.out.println("扣减库存失败，库存不足");
       }

   }finally { 
   
       // 释放锁
       redisTemplate.delete("lock");
   }

情况2：要是不抛异常，在释放锁之前服务重启了呢？那就来不及执行finally了吧，这下锁也没有释放吧。

那该怎么解决呢？

有小伙伴就会想：给锁加一个定时器嘛，它要挂就挂，到了时间锁自动释放，其他人又可以获取到锁服务又可用了

感觉是可行哦，那我们来实际操作一下。源码如下：

OK 定时有了，要是线程挂在里面，时间一到锁就会自动释放

事实真的会是这个样子的吗？

情况3：锁设置成功，接下来对锁进行定时，此时准备定时呢，还没定时成功突然程序挂了，又会导致死锁，像前那情况一样。

那这时我们就要保存设置锁和定时两个指令的原子性，要么全部成功要么全部失败，那该怎么实现呢？

这个问题大牛已经都想过了，往下看

Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "1213",10, TimeUnit.SECONDS);

一条更比两条强。这个底层使用 Lua 实现，保证其原子性。

再看一下我们优化后的代码

@GetMapping("/cut")
    public Object kc2() { 
   
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        /** * 设置锁，如果 lock 不存在的话，设置 lock=1213 并返回 true * 如果存在的话：就不操作 直接返回 false */
        String lockKey = "lock";
        String redisClientId = UUID.randomUUID().toString();
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockKey, redisClientId,10, TimeUnit.SECONDS);

        if(!lock){ 
   
            // 锁存在
            return "服务繁忙，请稍后再试";
        }
        try{ 
   
            int num = Integer.parseInt(redisTemplate.opsForValue().get("num").toString());
            if (num > 0) { 
   
                int lastNum = num - 1;
                redisTemplate.opsForValue().set("num", lastNum + "");
                System.out.println("扣减库存成功，剩余库存为：" + lastNum);
            } else { 
   
                System.out.println("扣减库存失败，库存不足");
            }
        }finally { 
   
            // 释放锁
            if (redisClientId.equals(redisTemplate.opsForValue().get(lockKey))){ 
   
                redisTemplate.delete(lockKey);
            }
        }
        return "ok";
    }

小伙伴会说，这下没问题啦哈哈。。。。

你确定没问题？

那就我来给你分析分析

正常逻辑：一个请求进来说去锁，设置有效时间为 10 秒，然后执行下面业务逻辑，最后释放锁，然后第二个线程进来。。。。

lock 有效时间为 10 秒，保不定那个线程执行任务的时候执行完要 15 秒，此时，lock 10秒就失效，那下一个线程就会进来，假如第二个线程要执行8 秒，第一个线程5秒后就执行完了然后释放lock 锁，线程1的锁早就失效了，它释放的锁确实线程2的锁，而第二个线程还有3秒才执行完，此时线程3获取到锁，又进来了，3秒后线程2又释放线程3 的锁，这样下去线程3释放线程4.。。。就会导致锁永久失效。

所以这个超时时间该设置多少呢？就需要根据项目来考量了。

那有什么好的解决方案吗，解决这个锁失效的问题呢？

有的，那肯定是有的，听我徐徐道来。。。

假设设置锁有效时间为 30 秒，那当线程获取锁后，开一个子线程做一个定时器，每隔一段时间去检查该对象的锁是否存在，存在的话就重新给锁续命。那如何保证自家的锁不会被别人释放呢？这下那个锁的 value 就派上用场了，给每个线程的锁配置上唯一标识（这个唯一标识就使用UUID）每次释放锁的时候就判断是否是自己的锁，保证只释放自己的锁。

理论有了，那该怎么实现呢？

使用Redisson 实现分布式锁

1、引入 Redisson 依赖

<dependency>
   <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.0</version>
</dependency>

2、配置Redisson

@Bean
public Redisson redisson(){ 
   
    // 此为单机模式
    Config config = new Config();
    config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0).setPassword("root");
    return (Redisson) Redisson.create(config);
}

3、代码实现

@Autowired
private Redisson redisson;

@GetMapping("/cut")
public Object kc() { 
   
    String lockKey = "lock";
    RLock rLock = redisson.getLock(lockKey);
    try { 
   
        // 加锁
        rLock.lock();
        int num = Integer.parseInt(redisTemplate.opsForValue().get("num").toString());
        if (num > 0) { 
   
            int lastNum = num - 1;
            redisTemplate.opsForValue().set("num", lastNum + "");
            System.out.println("扣减库存成功，剩余库存为：" + lastNum);
        } else { 
   
            System.out.println("扣减库存失败，库存不足");
        }
    } finally { 
   
        // 释放锁
        rLock.unlock();
    }
    return "ok";
}

但是主从架构也是可能出现问题，如 redis 挂了

后期优化可以用 redis 集群，增加高可用。

OK redis 锁分布式就讲完了，到这里锁就差不多了

要是还想更加完善可以选择使用ZK来实现，但是性能是没有 redis 好。

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/152027.html原文链接：https://javaforall.cn

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

云数据库 Redis

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度