实战系列-分布式锁的Redis实现

导语
本篇博客，博主使用本地的Docker 搭建了一套测试环境，用来手写一个属于自己的基于Redis分布式锁实现方案，通过自己实现来了解分布式锁的原理。并且对整个的构建过程做了分享，希望可以对大家学习分布式锁有所帮助。

文章目录

环境搭建
分布式锁需要解决的问题
- 互斥性
- 锁超时
- 阻塞或者非阻塞
- 可重入性
- 高可用
- Redission
- RedissonRedLock
总结

环境搭建

Redis
Spring Boot
Nginx
Jmeter

分布式锁需要解决的问题

互斥性
锁超时
支持阻塞与非阻塞
可重入性
高可用

互斥性

在一个电商系统中，如果一个客户进行下单操作，那么对应的要对实际的库存进行减扣操作，基于这样的一个场景，就会出现超卖的问题。那么什么是超卖问题，通过下面的演示来看一下什么是超卖问题？

@RestController
public class RedisController {private final static String product = "nihui";@Autowiredprivate RedisTemplate<String, Object> stringRedisTemplate;@GetMapping("/get")public void get() {// 超卖 问题// 这里实现了一个简单逻辑int stock = Integer.parseInt((String) stringRedisTemplate.opsForValue().get(product));if (stock > 0) {// 这里就开始执行下单的逻辑stock = stock - 1;stringRedisTemplate.opsForValue().set(product, String.valueOf(stock));System.out.println("进行库存的减扣 ， 现在库存 " + stock);} else {System.out.println("库存减扣失败，库存不足");}}
}

上面的代码逻辑是说，当我们访问/get 请求的时候说明用户进行了下单的操作，所以就要获取到当前的库存量，然后进行减一操作。将减一之后的结果更新到库存中。那么问题来了，在单用户操作的场景下没有任何问题，或者说再单线程的场景下没有任何问题，那么一个电商系统并不是简单的就支持一个用户，它需要支持的高并发。也就是说支持多个用户同时下单的操作。那么这个时候就会出现问题。下面就通过Jmeter来模拟

如图所示先测试一个量级比较小的，1秒中100个并发去访问这个接口。这里设置的默认商品大小是50个。开始测试一下结果。

从上面截图中可以看到库存为49 的时候，交易成功了很多订单，那么在一个电商系统中如果出现这样的问题。那么这个平台也不用干了，当然这个是开玩笑的，既然出问题了那么就要想办法解决啊？这个时候想到最多的就是多线程，线程安全的问题，是通过synchronized关键字来解决的。于是将我们的代码改成了如下的效果。

@GetMapping("/get")
public void get() {// 超卖 问题synchronized (product){// 这里实现了一个简单逻辑int stock = Integer.parseInt((String) stringRedisTemplate.opsForValue().get(product));if (stock>0){// 这里就开始执行下单的逻辑stock = stock - 1;stringRedisTemplate.opsForValue().set(product,String.valueOf(stock));System.out.println("进行库存的减扣 ， 现在库存 "+stock);}else {System.out.println("库存减扣失败，库存不足");}}
}

会看到整的库存减扣的操作被放到了一个synchronized关键字中，将所操作的商品进行了锁定操作。那么运行程序看看最后的测试结果如何。

通过结果观察看上去是没有什么问题的，但是真的没有问题么？这个是一个值得商榷的问题。在很多的电商网站，后台的应用程序并不是简单的一个应用程序来进行操作的。后台是挂载了很多的相同的应用程序服务，这些相同的应用程序组成的叫做集群，那么对于一个应用来说，并不是就只有一台机器。这个时候来模拟一个场景。在我们启动两台应用，并且通过Nginx将负载请求到两个机器上会出现什么样的结果？

同样的代码通过Nginx负载到了两个端口上下面是两个端口分别的结果
8080结果

8082结果

当然这里从两个结果上就可以看出，不难发现在8080端口上的结果有一个42，在8082端口的结果中也有一个42。这个只是在1秒100个并发，并且在两个实例中出现了这样的情况，如果并发量更大，实例数更多的场景中，那么损失将会是更大的。
这里就有人说在代码逻辑中不是已经加入synchronized关键字了么，为什么还会出现这样的问题。这里就要理解一下synchronized的原理，它是对于当前虚拟机而言的。对于两个实例它是运行到两个不一样的JVM中的。所以说它们两者之间是相互独立的，也就是不能保证两个独立的JVM之间加锁的。这个时候就需要一个公共的锁来提供服务。也就出现了分布式锁。通过分布式锁来解决保证互斥操作。

分布式锁原始模型

下面就来分析一下这个代码，首先我们知道，在一个代码执行过程中都是从上到下进行执行的，那么如果上面这段代码在执行删减库存操作的时候出现了问题，那么就会导致最后释放锁的逻辑没有被执行到。那么其他的线程进来访问这个操作的时候就会出问题。所以这里做的第一步优化是对这个业务逻辑的异常进行捕获。并且将释放锁的逻辑放入到finally中，如下

锁超时

那么我们知道finally中的代码一定会执行么，我想是不一定的，在Java中Exception是可以被捕获的。机房停电，kill -9 等操作，导致整个的逻辑还没有来得及执行finally就出现问题。也就导致了刚刚的错误。这个时候就会想到，其实在setnx操作之后还可以对这个所进行一个超时时间的设置。也就是说进行了如下的一个优化。

做完这个过期时间设置之后，即使在操作过程中出现了kill -9 这样的操作，也会在30秒之后自动的将锁进行释放，这样的话后续的操作线程还是可以获取到锁进行操作。那么这个时候这个代码还有没有其他问题呢？这里就涉及到Redis调用过程中的消耗问题。这段代码看上去没有问题，但是实际上，如果每一个线程进入之后都进行请求获取锁操作，也就是在RedisSession中存在大量的get set操作，而这些get set操作之间都是独立的，在一定程度上非常消耗IO和网络资源。也就是每个线程请求进入之后都要占用，如果进入的线程量大的话就会导致问题。那么这个改进方案呢就是在获取锁的同时就进行加入超时的操作。如下，看上去大功告成，但是实际上作为一个公共组件，最好是单独的进行处理操作。

作为一个公共组件来说，首先需要做的事情就是实现某种规则，那么这种规则如何实现，实现什么样的规则就需要使用到接口，对于一个锁来说，最主要的两个操作就是锁的获取，以及锁的释放操作。如下。

public interface Lock {public boolean getLock();public boolean releaseLock();
}

@Component
public class RedisLock implements Lock {private static final String lock = "lock";@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Overridepublic boolean getLock() {Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lock, lock,30,TimeUnit.SECONDS);return aBoolean;}@Overridepublic boolean releaseLock() {return stringRedisTemplate.delete(lock);}
}

这个操作完成之后就需要考虑一下了，在整个的操作过程中，多线程访问，怎么能保证这个线程进入之后和其他线程进入之后获取到的就是自己拿到的那把锁，或者说，这锁应该在整个的应用中是单例存在的。这样如何保证呢？因为我们知道在多线程执行过程中会涉及到一个指令重排序的操作，第一个线程进来之后正常的获取到锁了，注意这个是对Redis的操作，在Redis中设置了一个lock，那么其他线程进入的时候其实已经开始准备执行release的操作了那么这个时候就会导致在Redis不存在lock，后续的线程就会进行重新的设置。导致整个系统的死锁，或者是服务假死。那么就要保证当前线程操作当前锁。这里先做了如下的一个操作

首先从代码逻辑的角度上来说好像是可以保证了当前线程操作当前锁，那么带来的另外的问题就是这个UUID会被覆盖，这样就出现了线程安全问题，有人就说了可不可以考虑用一个不被覆盖的内容来进行操作呢？线程安全并且还可以在随着线程传递点东西，那么首选的就是ThreadLocal。下面对代码进行如下的改进。

上面这个代码就保证了，其他线程不会拿到当前线程的UUID，也就不会对UUID产生覆盖了。既然是这样的操作，可以保证了线程安全。那么会不会有其他问题呢？我们知道在使用ThreadLocal的时候在第一个线程使用过之后，如果没有进行清理，当复用线程的时候在当前线程的ThreadLocal中就会有脏数据。这个时候就会出现其他线程也可以对锁进行释放。这个时候就需要将整个的ThreadLocal在使用完成之后进行remove操作。

到这里看上去我们的分布式锁应该是没有问题了，那么就要结合正常的使用场景来进行分析了

阻塞或者非阻塞

在这里可以看到如果后续线程没有获取到锁之后会直接进行返回操作，在一定场景下并不是太满足场景业务要求。那么就需要进行一个优化操作，实现一个阻塞与非阻塞的调配。例如一个抢购的场景，抢购的商品就只有100个，而正好从各个地方过来的请求也只有100个，那么当第一个人下单的时候，这个时候其他人获取锁的时候都是false，意味着后进入的99个用户其实是没有买到商品的。那么这个时候就需要让锁支持一个阻塞操作。也就是说如果后续没有拿到锁，就一直尝试获取这个锁，而不是直接返回。对于直接返回的操作就可以看作是一个非阻塞的，而对于等待获取就是一个阻塞的。

这里就需要对分布式锁组件可以进行阻塞操作。这里采用了自旋的方式来实现。也就是先在外面定义一个锁标识，如果这个锁标识为获取到锁则直接返回，如果没有获取到锁则进入到一个死循环中，一直尝试去获取这个锁，直到获取到锁之后跳出本次的循环

对于阻塞与非阻塞在很多场景中需要做一定的取舍，因为在阻塞状态下的死循环其实是一个非常消耗CPU内存的一个操作。所以要在这个地方做一个优化。

可重入性

在有些业务场景中，存在这样一个场景，就是A调用B，B调用C，而在使用C服务的时候已经拿到了着锁，这个时候就会导致当前线程操作无法获取到这个锁，但是实际上从某种角度上讲，C服务的操作与当前操作是有关联的，我们需要让这个锁被当前服务也可以获取到

这个时候上面这种场景就需要支持这个锁的可重用，类似于ReentrantLock。为什么会有这个内容呢？就是因为在使用synchronized关键字的时候，在外部进行加锁操作之后，在内部继续使用synchronized关键字进行加锁，这个内部的锁是不起作用的。从上面的调用可以看到，在整个的业务逻辑中其实并不是这么简单的操作，如果使用这样的一个锁操作，导致整个的C获取不到锁，导致程序无法继续执行。那么就需要对锁操作进行优化。通过ThreadLocal的UUID标识判断来获取当前是否是第一次获取锁，如果是则执行新逻辑，如果不是则直接返回已经获取过锁，保证了锁的重入性。

高可用

如果第一个请求过来了开始获取锁，所有的对于锁的请求问题都已经解决了，并且设置了30秒的过期时间，但是这里一定可以保证这个锁拿到之后，后续的业务操作就一定是低于30秒进行返回么，如果超过30秒那么就导致在释放锁的时候没有该锁这个一个问题，30秒之后这个锁失效了，其他线程就会继续进入到该操作中，但是实际上当前操作并没有完成。那么这样这个问题怎么解决？其实这个就是可以看做一个锁穿透，几乎可以看错一个操作超时导致其他操作超时，整个所有的锁都是不起作用的。但是如果这个时间设置太长的话，就会影响整个的系统性能。这个时候就需要在低性能的情况下达到Redis分布式锁的高可用。继续对这个问题进行优化

上面代码中可以看到在获取锁的过程中加入了一个while的死循环，也就是说在获取完之后每隔10秒为这个锁做一个自动的延期操作，但是实际上这地方是有问题的，没有对后续的获取锁操作进行释放，但是这个延期操作，在这个锁存在的过程中有必须得一直进行，这个时候就需要考虑到用多线程的来解决。也就是做一个异步操作

做完异步操作之后需要在释放的时候把这个线程停止，那么怎么实现呢？这个时候需要考虑一个问题，线程与子线程的问题。这里会看到，如果将调用组件的线程看作是父线程，那么新创建的这个线程应该是属于它的子线程，也就是说通过某种手段，在不影响父线程的情况下把子线程给做停止操作。这里手段很多，这里借助于ThreadLocal来标识是当前线程的子线程。

@Component
public class RedisLock implements Lock {private static final String lock = "lock";private ThreadLocal<String> stringThreadLocal = new ThreadLocal<>();@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Overridepublic boolean getLock() {boolean lockTemp = false;//如果当前没有获取过锁则进入新获取锁的逻辑if (stringThreadLocal.get()==null){Thread thread = new Thread(){@Overridepublic void run() {while (true){//这里的操作就是不断的执行当前锁每隔10秒设置一个过期时间，这个过期时间就是30秒stringRedisTemplate.expire(lock,30,TimeUnit.SECONDS);try {TimeUnit.SECONDS.sleep(100000);} catch (InterruptedException e) {e.printStackTrace();}}}};String uuid= thread.getId()+":"+UUID.randomUUID().toString();stringThreadLocal.set(uuid);lockTemp = stringRedisTemplate.opsForValue().setIfAbsent(lock, uuid,30,TimeUnit.SECONDS);if (!lockTemp){while (true){lockTemp = stringRedisTemplate.opsForValue().setIfAbsent(lock, uuid,30,TimeUnit.SECONDS);if (lockTemp){break;}}}thread.start();}else if (stringThreadLocal.get().equals(stringRedisTemplate.opsForValue().get(lock)){return true;}return lockTemp;}@Overridepublic boolean releaseLock() {String uuid = stringThreadLocal.get();Integer id = Integer.parseInt(uuid.split(":")[0]);Thread thread = findThread(id);thread.stop();if (uuid.equals(stringRedisTemplate.opsForValue().get(lock))){stringRedisTemplate.delete(lock);stringThreadLocal.remove();return true;}return false;}public static Thread findThread(long threadId){ThreadGroup group = Thread.currentThread().getThreadGroup();while (group !=null){Thread[] threads = new Thread[(int)(group.activeCount()*1.2)];int count = group.enumerate(threads,true);for (int i = 0; i < count; i++) {if (threadId == threads[i].getId()){return threads[i];}}group = group.getParent();}return null;}
}

到这里整个的分布式的逻辑就完成了。当然这里使用自己编写的方式来实现分布式锁，但是在很多的框架中都有所实现，这里比较常用的就是RedisSession

Redission

这里来看看RedisSession是怎么实现分布式锁的，来比较一下两者的实现在考虑问题上有什么样差距。引入Maven依赖

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

进入到lock方法之后会看到如下的代码

从这里可以看到调用的是一个script并且使用了Future模式。也就是说我们等待，或者期待给我们返回一个RedissonLockEntry。

RedissonRedLock

这里提出一个概念，使用这个锁可以高效的保证高可用，因为使用它的时候所请求的是不同的实例，只有当这些实例半数以上返回成功的时候才会进行加锁操作。当然如果没有具体的业务场景，看上去这些操作都是虚无缥缈的。所以这些操作还是要结合具体的业务场景。

总结

本篇博客手撕了一下分布式锁的实现，并且自己通过具体搭建测试环境来实现了一个有待优化的分布式锁，当然这个分布式锁的优化还有很多的地方，要想实现更加高效的的分布式锁，就需要从最底层的代码开始进行优化。这里有兴趣的读者可以自己研究研究。