【📕分布式锁通关指南 02】基于Redis实现的分布式锁
原创【📕分布式锁通关指南 02】基于Redis实现的分布式锁
原创
引言
在01篇文章中，我们深入探讨了单机锁的多种实现方式，并相信各位读者已经对它们有了较为全面的了解。然而，随着我们对单机锁的深入了解，不难发现它们所固有的一些局限性。因此，从本篇开始，我们将开始探讨分布式锁的相关内容。
认识分布式锁
首先，先来看它的概念-控制分布式系统之间同步访问共享资源的一种方式。所以，它需要满足以下四个特性：互斥性、可重入性、锁超时防死锁、锁释放正确防误删。而01篇中提到的JVM锁在分布式场景中就会存在问题，比如，我们当前有两个服务实例，它们都访问商品库存表进行扣减库存，如果使用JVM锁，其实并没有效果，如图：
JVM锁只能锁所在服务的实例，所以在分布式场景下，有多少个服务实例自然也会存在多少个JVM锁。那么有解决办法吗？当然是有的。没有什么是加一层解决不了的，我们只需要在服务实例和数据库之间再加一层作为分布式锁即可，如图：
我们可以依靠中间件来实现加的这一层，常见的有reids、Zookeeper、Etcd等，本篇我们将以redis分布式锁的实现展开讲解，其他实现也会在后续篇中陆续讲解。
redis实现分布式锁的思路
在开始实现前，我们先来聊聊为什么选择redis来实现分布式锁。这里做技术选型，自然离不开对中间件本身的特点进行分析，redis的以下特点足够支持它来实现分布式锁：
- 1.Redis是高性能的内存数据库，满足高并发的需求；
- 2.Redis支持原子性操作，保证操作的原子性和一致性；
- 3.Redis支持分布式部署，支持多节点间的数据同步和复制，从而满足高可用性和容错性。
除了上述特性，redis客户端提供的一个命令让我们设置锁也变得更为简单，即setnx，区别于set命令，使用它来设置键值对，如果键已存在，就不会设置成功。所以使用这个命令来获取锁的话，我们可以省去很多判断逻辑。
redis实现简化版分布式锁
有了思路，我们可以尝试用代码来实现下。首先，使用redisTemplate来实现下加锁和解锁的方法。加锁就是用setnx命令设置个键值对,key根据业务场景设置,value随意；解锁就是根据key删除指定的键值对，如下：
@Override
public void lock() {
//1.使用setnx指令进行加锁
while (true) {
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, "1");
if (result != null && result) {
break;
}
}
}
@Override
public void unlock() {
stringRedisTemplate.delete(this.lockName);
}接着我们继续以扣减库存为例，大致逻辑应该是先获取锁，锁的key就是商品id，拿到锁之后先判断库存数量是否足够，如果足够，则去扣减库存。如下：
public String deductStockRedisLock(Long goodsId,Integer count) {
AbstractLock lock = null;
try {
lock = new RedisLock(template, "stock" + goodsId);
lock.lock();
//1.查询商品库存数量
String stock = template.opsForValue().get("stock" + goodsId);
if (StringUtil.isNullOrEmpty(stock)) {
return "商品不存在！";
}
int lastStock = Integer.parseInt(stock);
//2.判断库存数量是否足够
if (lastStock < count) {
return "库存不足！";
}
//3.如果库存数量足够，则去扣减库存
template.opsForValue().set("stock" + goodsId, String.valueOf(lastStock - count));
return "扣减库存成功";
} finally {
if (lock != null) {
lock.unlock();
}
}
}接着我们启动熟悉的JMeter来进行测试，在开始前，我们先往redis里set一个key为stock1，value为6000的键值对来表示id为1的商品有6000库存，如下：
启动JMeter观察执行报告，会发现吞吐量很低，这里读者可以自行对比01篇中的数据。最直接的体现就是这里的扣减库存执行了差不多20s左右才完成，如下：
这个执行效率如果放到线上肯定是不行的，前面也讲过我们选择redis是奔着高性能去的，可是为什么表现却这么差呢？我们看下加锁的逻辑，如下：
public void lock() {
//1.使用setnx指令进行加锁
while (true) {
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, "1");
if (result != null && result) {
break;
}
}
}我们这里的加锁逻辑是只要没获取到锁就去重试，而redis的写命令执行的也比较快，所有这里在高并发场景下就变成了低效重试，那么有没有解决办法呢？当然是有的，很简单，我们只需要在获取失败后，让当前线程先停一下即可，如下：
public void lock() {
//1.使用setnx指令进行加锁
while (true) {
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, "1");
if (result != null && result) {
break;
}
try {
Thread.sleep(50);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}简化版分布式锁存在的问题
在上面的代码里，我们基于redis手撸了一个简化版的分布式锁，那么它是否就满足日常业务使用了呢？当然不行，既然是简化版的自然就存在问题。我们先来分析一下前文中提到的四个特性中的其中两个-锁超时防死锁和锁释放正确防误删，那么我们的简化版能否满足呢？显然是不行的，因此就需要我们继续迭代了。
1.锁超时怎么办？
锁超时的情况可能有很多，比如扣减库存获取锁之后代码执行到一半服务挂掉了，由于是异常关闭，所以finally中释放锁的逻辑也没来得及执行，这个时候锁就被永久的持有了。所以为了解决这个问题，我们就需要为锁加上过期时间，这样可以保证无论业务或者服务是否出现异常，最终都可以保证锁的释放，代码如下：
private final long defaultExpireTime = 30000;
@Override
public void lock() {
lock(TimeUnit.MILLISECONDS, defaultExpireTime);
}
@Override
public void lock(TimeUnit timeUnit, Long expireTime) {
//1.使用setnx指令进行加锁
while (true) {
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, "1", expireTime, timeUnit);
if (result != null && result) {
break;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}所以其实很简单，只需要给锁加个过期时间就可以了，这个时间根据自己的业务场景定。因为如果你定的少了，假如我们定的过期时间是500毫秒，但是相应的业务逻辑执行完成需要800毫秒，那么就会造成业务逻辑还没执行完成，锁就被释放了，这锁就是加了个寂寞。
2.锁被误删了怎么办？
首先，我们来定义下什么叫锁误删，即某个线程持有的锁被别的线程删了。那么这里肯定就有同学疑惑了，按照我们上面的代码逻辑，假设现在有个A线程获取到锁了，在它没释放的情况下，其他线程应该是一直循环获取才对，也就是说这个时候其他线程根本就拿不到这把锁，又怎么能给它释放了呢。
其实问题就出在我们上面为了解决锁超时问题而给锁加了过期时间，我们假设A线程的业务逻辑处理的时间超过了锁超时释放的时间，就造成了A线程还没执行完，锁就自己释放了，这个时候B线程获取到了锁开始执行，而A线程继续执行到了释放锁的逻辑。注意：此时按照我们的设计，锁的key是商品id，也就是说A、B两线程拿到的是同一把锁，那么这个时候A线程的释放锁反而把B线程拿到的给释放了，最终肯定会造成并发问题的。那么知道了问题所在，我们怎么解决呢？很简单，只需要在释放锁之前判断下当前释放锁的线程是否是拿到锁的线程不就好了，只有一致的情况下才可以释放锁，代码如下：
@Override
public void lock(TimeUnit timeUnit, Long expireTime) {
//1.使用setnx指令进行加锁
while (true) {
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockName, uuid, expireTime, timeUnit);
if (result != null && result) {
break;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
@Override
public void unlock() {
//1.判断当前持有锁线程是否等于本线程
String result = stringRedisTemplate.opsForValue().get(this.lockName);
if (this.uuid.equals(result)) {
stringRedisTemplate.delete(this.lockName);
}
}我们这里的做法是在获取锁的时候给value设置一个uuid，并在删除之前先判断当前线程的uuid和锁对应的uuid是否一致。
小结
本章节通过redis实现了一套简易的分布式锁，看似我们现在的设计已经非常完美，解决了锁超时和锁误删的问题，但实际上还有一些问题没有解决，比如释放锁那里，如果线程A执行过判断后刚好到了锁自然释放的时间，于是释放掉了，而正要执行删除锁的时候，线程B已经拿到锁了，但此时线程A肯定也不知道uuid已经发生变化了，于是执行删除顺利地把线程B刚拿到的锁给释放了，顺利地造成了后续的并发问题。因此，我们将在下一章解决这样的问题。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
![[Redis] 分布式缓存中间件 Redis 之 分布式锁实战](https://ask.qcloudimg.com/http-save/yehe-5885900/9fwvmof2ct.jpeg)
腾讯云开发者
