快播：一文读懂分布式锁——使用SpringBoot+Redis实现分布式锁

随着现在分布式架构越来越盛行，在很多场景下需要使用到分布式锁。很多小伙伴对于分布式锁还不是特别了解，所以特地总结了一篇文章，让大家一文读懂分布式锁的前世今生。

分布式锁的实现有很多种，比如基于数据库、Redis 、 zookeeper 等实现，本文的示例主要介绍使用Redis实现分布式锁。

一、什么是分布式锁

分布式锁，即分布式系统中的锁，分布式锁是控制分布式系统有序的对共享资源进行操作，在单体应用中我们通过锁实现共享资源访问，而分布式锁，就是解决了分布式系统中控制共享资源访问的问题。

可能初学的小伙伴就会有疑问，Java多线程中的公平锁、非公平锁、自旋锁、可重入锁、读写锁、互斥锁这些都还没闹明白呢？怎么又出来一个分布式锁？

其实，可以这么理解：Java的原生锁是解决多线程下对于共享资源的操作，而分布式锁则是多进程下对于共享资源的操作。分布式系统中竞争共享资源的最小粒度从线程升级成了进程。

分布式锁经被应用到各种高并发的场景下，典场景案例包括：秒杀、车票、订单、退款、库存等场景。

二、为什么要使用分布式锁

在分布式系统中，常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，这个时候，便需要使用到分布式锁。

目前几乎很多大型网站及应用都是分布式部署的，如何保证分布式场景中的数据一致性问题一直是一个比较重要的话题。在某些场景下，为了保证数据的完整性和一致性，我们需要保证一个方法在同一时间内只能被同一个线程执行，这就需要使用分布式锁。

如上图所示，假设用户A和用户B同时购买了某款商品，订单创建成功后，下单系统A和下单系统B就会同时对数据库中的该款商品的库存进行扣减。如果此时不加任何控制，系统B提交的数据更新就会覆盖系统A的数据，导致库存错误，超卖等问题。

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

在介绍分布式锁的实现方式之前，先了解一下分布式锁应该具备哪些条件：

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

2、高可用的获取锁与释放锁；

3、高性能的获取锁与释放锁；

4、具备可重入特性；

5、具备锁失效机制，防止死锁；

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

四、分布式锁的实现方式

随着业务发展的需要，原来的单体应用被演化成分布式集群系统后，由于系统分布在不同机器上，这就使得原有的并发控制锁策略失效，为了解决这个问题就需要一种跨进程的互斥机制来控制共享资源的访问，这就需要用到分布式锁！

分布式锁的实现有多种方式，下面介绍下这几种分布式锁的实现：

基于数据库实现分布式锁，(适用于并发小的系统)；

基于缓存（Redis等）实现分布式锁，（效率高，最流行，存在锁超时的问题）；

基于Zookeeper实现分布式锁，（可靠，但是效率不搞）；

尽管有这三种方案，但是不同的业务也要根据自己的情况进行选型，他们之间没有最好只有更适合！

五、基于Redis实现分布式锁

使用Redis实现分布式锁是目前比较流行的解决方案，主要是使用Redis 获取锁与释放锁效率都很高，实现方式也特别简单。

实现原理：

（1）获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断。

（2）获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁。

（3）释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放。

接下来我们就一步一步实现Redis 分布式锁。

第一步，创建Spring Boot项目，并引入相关依赖。

相关资料图

org.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-data-redisorg.springframework.bootspring-boot-starterorg.springframework.bootspring-boot-starter-testtestorg.apache.commonscommons-lang3com.alibabafastjson1.2.72org.projectlomboklombok

第二步，创建Redis分布式锁通用操作类，示例代码如下：

@Slf4j@ComponentpublicclassRedisLockUtils{@ResourceprivateRedisTemplateredisTemplate;privatestaticMaplockInfoMap=newConcurrentHashMap<>();privatestaticfinalLongSUCCESS=1L;publicstaticclassLockInfo{privateStringkey;privateStringvalue;privateintexpireTime;//更新时间privatelongrenewalTime;//更新间隔privatelongrenewalInterval;public static LockInfo getLockInfo(String key, String value, int expireTime) {LockInfo lockInfo = new LockInfo();lockInfo.setKey(key);lockInfo.setValue(value);lockInfo.setExpireTime(expireTime);lockInfo.setRenewalTime(System.currentTimeMillis());lockInfo.setRenewalInterval(expireTime * 2000 / 3);return lockInfo;}public String getKey() {return key;}public void setKey(String key) {this.key = key;}public String getValue() {return value;}public void setValue(String value) {this.value = value;}public int getExpireTime() {return expireTime;}public void setExpireTime(int expireTime) {this.expireTime = expireTime;}public long getRenewalTime() {return renewalTime;}public void setRenewalTime(long renewalTime) {this.renewalTime = renewalTime;}public long getRenewalInterval() {return renewalInterval;}public void setRenewalInterval(long renewalInterval) {this.renewalInterval = renewalInterval;}}/*** 使用lua脚本更新redis锁的过期时间* @param lockKey* @param value* @return 成功返回true, 失败返回false*/publicbooleanrenewal(StringlockKey,Stringvalue,intexpireTime){String luaScript = "if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("expire", KEYS[1], ARGV[2]) else return 0 end";DefaultRedisScript redisScript = new DefaultRedisScript<>();redisScript.setResultType(Boolean.class);redisScript.setScriptText(luaScript);List keys = new ArrayList<>();keys.add(lockKey);Object result = redisTemplate.execute(redisScript, keys, value, expireTime);log.info("更新redis锁的过期时间：{}", result);return (boolean) result;}/*** @param lockKey    锁* @param value      身份标识（保证锁不会被其他人释放）* @param expireTime 锁的过期时间（单位：秒）* @return 成功返回true, 失败返回false*/public boolean lock(String lockKey, String value, long expireTime) {returnredisTemplate.opsForValue().setIfAbsent(lockKey,value,expireTime,TimeUnit.SECONDS);}/*** redisTemplate解锁* @param key* @param value* @return 成功返回true, 失败返回false*/public boolean unlock2(String key, String value) {Object currentValue = redisTemplate.opsForValue().get(key);boolean result = false;if (StringUtils.isNotEmpty(String.valueOf(currentValue)) && currentValue.equals(value)) {result = redisTemplate.opsForValue().getOperations().delete(key);}return result;}/*** 定时去检查redis锁的过期时间*/@Scheduled(fixedRate = 5000L)@Async("redisExecutor")public void renewal() {long now = System.currentTimeMillis();for (Map.Entry lockInfoEntry : lockInfoMap.entrySet()) {LockInfo lockInfo = lockInfoEntry.getValue();if (lockInfo.getRenewalTime() + lockInfo.getRenewalInterval() < now) {renewal(lockInfo.getKey(), lockInfo.getValue(), lockInfo.getExpireTime());lockInfo.setRenewalTime(now);log.info("lockInfo {}", JSON.toJSONString(lockInfo));}}}/*** 分布式锁设置单独线程池* @return*/@Bean("redisExecutor")public Executor redisExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(1);executor.setMaxPoolSize(1);executor.setQueueCapacity(1);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("redis-renewal-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());return executor;}}

第三步，创建RedisTemplate 配置类，配置Redistemplate,示例代码如下：

@ConfigurationpublicclassRedisConfig{@Beanpublic RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) throws Exception {RedisTemplate redisTemplate = new RedisTemplate();redisTemplate.setConnectionFactory(redisConnectionFactory);// 创建 序列化类GenericToStringSerializer genericToStringSerializer = new GenericToStringSerializer(Object.class);redisTemplate.setKeySerializer(new StringRedisSerializer());redisTemplate.setValueSerializer(genericToStringSerializer);return redisTemplate;}}

第四步，实现业务调用，这里以扣减库存为例，示例代码如下：

@RestControllerpublic class IndexController {@Resourceprivate RedisTemplate redisTemplate;@Autowiredprivate RedisLockUtils redisLock;@RequestMapping("/deduct-stock")public String deductStock() {String productId = "product001";System.out.println("---------------->>>开始扣减库存");String key = productId;String requestId = productId + Thread.currentThread().getId();try {boolean locked = redisLock.lock(key, requestId, 10);if (!locked) {return "error";}//执行业务逻辑//System.out.println("---------------->>>执行业务逻辑:"+appTitle);int stock = Integer.parseInt(redisTemplate.opsForValue().get("product001-stock").toString());int currentStock = stock-1;redisTemplate.opsForValue().set("product001-stock",currentStock);try {Random random = new Random();Thread.sleep(random.nextInt(3) *1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("---------------->>>扣减库存结束:current stock:" + currentStock);return "success,current stock:" + currentStock;} finally {redisLock.unlock2(key, requestId);}}}

六、验证测试

代码完成之后，开始测试。我们同时启动两个实例，端口号为：8888和8889模拟分布式系统。

接下来，我们分别请求：http://localhost:8888/deduct-stock

和http://localhost:8889/deduct-stock，或者使用JMater分别请求这两个地址，模拟高并发的情况。

通过上图我们可以看到，在批量请求的情况下，库存扣减也没有出现问题。说明分布式锁生效了。

最后

以上，我们就把什么是分布式锁，如何基于Redis 实现分布式锁的解决方案介绍完了。分布式锁是分布式系统中的重要功能组件，希望大家能够熟练掌握。

关键词： 分布式系统 解决方案