分布式锁是控制不同系统之间访问共享资源的一种锁实现，如果不同的系统或同一个系统的不同主机之间共享了某个资源时，往往需要互斥来防止彼此干扰来确保数据一致性。
在单机部署的系统中，一般采用线程锁来解决高并发的问题，多线程访问共享变量的问题达到数据一致性，可以使用synchornized、ReentrantLock等。但是对于分布式集群部署的系统架构，程序在不同的JVM虚拟机中运行，且因为synchronized或ReentrantLock都只能保证同一个JVM进程中保证有效，所以这时就需要使用分布式锁了。
目前主流的分布式锁的实现方式：基于数据库的分布式锁、基于Redis实现的分布式锁、基于zookeeper实现分布式锁 三种，本文主要介绍采用Redis的方式实现分布式锁。
以优惠券拼抢的场景为说明，一般的话在818或者其他购物节到来之前，系统都会预先发放对应的优惠券，假设在某一天，系统放发乐一个满减优惠券couponTicket，发放数量为50:，通过伪代码的方式模拟一下多线程抢券的场景：
/** * 营销优惠券 * * @author: jacklin * @date: 2022/8/17 14:58 */@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } return "end"; }}复制代码代码思考分析：不难发现，以上的代码很明显存在一个问题，如果在同一时间两个线程thread1、thread2同时请求进来，通过redisTemplate.opsForValue().get(lockKey)拿到优惠券剩余量都是50，当执行完成后thread1和thread2又将减完后的库存couponTicket=49重新写入Redis，那么数据就会产生问题，实际上两个线程都抢券成功，实现应该各减去了一张券码数，然而实际写进就只减了一次券码数，就出现了数据不一致的现象。
上述的问题产生的原因主要就是从Redis拿数去到扣减优惠券数量不是原子性操作，在以往的单体项目中，解决这种问题可以采用synchronized同步锁的方式去实现：
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; synchronized (this) { int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } return "end"; }}复制代码代码思考分析：当多线程并发访问的时候，执行到synchronized(this)位置的时候，只会有一个线程获得锁，进入synchronized代码块中执行业务逻辑，当该线程执行完成之后才会释放锁，等下一个线程进来又会重新上锁执行该段代码，简单来说，通过synchronized锁机制实现多线程排队执行代码块的代码，从而保证了线程的安全。
以上的做法，如果是后端服务仅部署在一台机器的情况下（也就是单机环境），这样的实现是没有问题的，但是在如今的互联网公司，面对这些高并发的场景，后端服务肯定是不仅部署到服务器上的，最好都有两台甚至更多机子来构建集群的架构，那么采用synchronized加锁的方式很明显解决不了当前的资源访问控制的问题了。可以通过jmeter去模拟并发请求，在集群模式并发场景下，还是会出现数据不一致的问题。
若后端是两个微服务构成的服务集群，由于synchronize代码块只能在同一个JVM进程中生效，两个请求能够同时进两个服务，所以上面代码中的synchronized就一点作用没有了。synchronized和juc包下个那些锁都是只能用于JVM进程维度的锁，并不能运用在集群或分布式部署的环境中。
通过上面的实验很容易发现通过synchronized等JVM进程级别的锁并不能解决分布式场景的并发问题，而正因为需要解决这种场景下的问题，才有了分布式锁的出现。
可以通过Redis的SETNX命令（只在键key不存在的情况下，将键key的值设置为value。若键key已经存在，则SETNX命令不做任何动作）来解决，主要解决上面 集群环境下锁不唯一的问题。
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; // redis setnx操作 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "123456"); if (Boolean.FALSE.equals(result)) { return "error"; } int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } redisTemplate.delete(lockKey); return "end"; }}复制代码代码思考分析：看到上述代码的实现，通过redis setnx分布式锁的实现，实际上该段代码还是存在问题的，就是当执行扣减余票操作时，若业务代码报了异常，那么就会导致后面的删除Redis的key代码没有执行到，就会使Redis的key没有删掉的情况，那么Redis的这个key就会一直存在Redis中，后面的线程再进来执行下面这行代码都是执行不成功的，就会 导致线程死锁，会导致后面进来的线程一直拿不到锁，一直请求不到资源，那么问题就会很严重了。
为了解决上述问题其实很简单，只要加上一个try...finally即可，这样业务代码即使抛了异常也可以正常的释放锁。setnx + try ... finally解决，具体代码如下：
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; try { // redis setnx操作 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "123456"); if (Boolean.FALSE.equals(result)) { return "error"; } int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } catch (Exception e) { e.printStackTrace(); } finally { //释放锁 redisTemplate.delete(lockKey); } return "end"; }}复制代码代码思考分析：上述业务代码执行报错的问题解决了，但是又会有新的问题，当程序执行到try代码块中某个位置服务宕机或者服务重新发布，这样就还是会有上述的Redis的key没有删掉导致死锁的情况。这样可以使用Redis的过期时间来进行设置key，setnx + 过期时间解决，如下代码所示：
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; try { // redis setnx操作 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "123456"); //设置过期时间 redisTemplate.expire(lockKey, 3, TimeUnit.SECONDS); if (Boolean.FALSE.equals(result)) { return "error"; } int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } catch (Exception e) { e.printStackTrace(); } finally { //释放锁 redisTemplate.delete(lockKey); } return "end"; }}复制代码代码思考分析：上述代码解决了因为程序执行过程中宕机导致的锁没有释放导致的死锁问题，但是如果代码像上述的这种写法仍然还是会有问题，当程序执行到第 redisTemplate.opsForValue().setIfAbsent(lockKey, "austin");行时，系统突然宕机了，此时Redis的过期时间并没有设置，也会导致线程死锁的现象。可以用了Redis设置的原子命设置过期时间的命令，原子性过期时间的setnx命令，如下代码所示：
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; try { // redis setnx + 过期时间 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, "123456",3,TimeUnit.SECONDS); if (Boolean.FALSE.equals(result)) { return "error"; } int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } catch (Exception e) { e.printStackTrace(); } finally { //释放锁 redisTemplate.delete(lockKey); } return "end"; }}复制代码代码思考分析：通过设置原子性过期时间命令可以很好的解决上述这种程序执行过程中突然宕机的情况。这种Redis分布式锁的实现看似已经没有问题了，一般软件公司并发量不是很高的情况下，这种实现分布式锁的方式已经够用了，即使出了些小的数据不一致的问题，也是能够接受的，但是 如果是在高并发的场景下，上述的这种实现方式还是会存在很大问题。
这里我们举个简单说明一下该方式存在的缺陷问题：
线程A抢占了锁，并设置了这个锁的过期时间为10秒，也就是10秒后自动开锁，锁的编号为123456。
10秒以后，线程A业务还在执行，此时锁就被自动打开了。
线程B进来发现锁已经被打开了，于是抢占了锁，设置锁的编号为123456，并设置锁的过期时间为10秒。 线程A在15秒的时候，完成了任务，但此时线程B还在执行任务。
线程A主动打开了编号为123456的锁，刚好这个锁也是线程B的锁。。
线程B还在执行任务，发现自己的锁已经被打开了。。
线程B：我都还没执行完任务呢，为什么我的锁开了？
线程B的锁在线程A执行完的时候释放了，此时，线程B还在执行任务。 线程C抢占到了锁，线程C开始执行任务 此时出现线程B和线程C同时执行一个任务，在执行任务上产生了冲突。
从上面的分析可以知道，当当前线程处理 执行业务逻辑所需要的时间大于锁的过期时间 ，这时候锁会自动释放，又会被其他线程抢占到锁，当前线程执行完之后，会把其他线程抢占到的锁误释放。然而，为什么会误打开别的线程的锁呢？因为锁的唯一性，每个锁的编号都是123456，线程只认锁的编号，看见编号为123456的就开，结果把别人的锁打开了，这种情况就是锁 没确保唯一性导致的。
解决上述问题其实也很简单，让每个线程加的锁时给Redis设置一个唯一id的value，每次释放锁的时候先判断一下线程的唯一id与Redis 存的值是否相同，若相同即可释放锁。设置线程id的原子性过期时间的setnx命令， 具体代码如下：
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; String threadUniqueKey = UUID.randomUUID().toString(); try { // redis setnx + 唯一编号 + 过期时间 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, threadUniqueKey, 3, TimeUnit.SECONDS); if (Boolean.FALSE.equals(result)) { return "error"; } int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } catch (Exception e) { e.printStackTrace(); } finally { //释放锁 redisTemplate.delete(lockKey); } return "end"; }}复制代码代码思考分析：实际上述实现的Redis分布式锁已经能够满足大部分应用场景了，但是还是略有不足，比如当线程进来需要的执行时间超过了Redis key的过期时间，此时锁就已经被释放了，其他线程就可以立马获得锁执行代码，就又会产生bug了。
分布式锁的key过期时间不管设置成多少都不合适，比如设置为10秒，如果业务代码执行链路较长，亦或是代码存在慢SQL、数据查询量大的情况，那么过期时间就不好设置，那么这里有没有什么更好的方案呢？答案是有的：锁续命。
那么锁续命方案的原来就在于当线程加锁成功时，会开一个分线程，取锁过期时间的1/3时间点定时执行任务，每10s判断一次锁是否存在(即Redis的key)，若锁还存在那么就直接重新设置锁的过期时间，若锁已经不存在了那么就直接结束当前的分线程。
上面说的续命锁看起来简单，但是实际上实现还是有一定的难度的，于是类似 Redission 开源框架已经帮我们实现好了，所以不需要再重复造轮子自己去写一个分布式锁了，下面会拿Redission框架举例，学习一下Redission分布式锁的设计思想。
Redission实现分布式锁的原理流程图如下图所示，当线程一加锁成功获取到锁，并开启执行业务代码时，Redission框架会开启一个后台线程（所谓的Watch Dog看门狗），每隔锁过期的1/3时间去定时判断一次是否还持有锁（Redis的key是否还存在），若不持有那么久执行结束当前的后台线程，若还持有锁，那么久重新设置锁的过期时间，当线程一加锁成功后，那么线程二就自然获取不到锁，此时线程二就会做类似CAS的自旋操作，一直等待线程一释放锁之后才能加锁成功。
另外，Redison底层实现分布式锁时使用了大量的lua脚本保证了其加锁操作的各种原子性。Redison实现分布式锁使用lua脚本的好处主要是能保证Redis的操作是原子性的，Redis会将整个脚本作为一个整体执行，中间不会被其他命令插入。
@Slf4j@RestControllerpublic class MarketCouponController { @Resource private Redisson redisson; @Resource private StringRedisTemplate redisTemplate; /** * Redission分布式锁实现 * * @author: jacklin * @since 2022/8/18 11:25 **/ @RequestMapping("/deductCouponTicket") public String deductCouponTicket(String couponCode) { String lockKey = "coupon:ticket" + couponCode; RLock rlock = redisson.getLock(lockKey); try { rlock.lock(); int couponTicketCount = Integer.parseInt(redisTemplate.opsForValue().get(lockKey)); if (couponTicketCount > 0) { int realCouponTicketCount = couponTicketCount - 1; log.info("用户抢券成功，扣减优惠券数量，剩余量：" + realCouponTicketCount); redisTemplate.opsForValue().set(lockKey, realCouponTicketCount + ""); } else { log.error("用户抢券失败，优惠券已被抢光！"); } } catch (Exception e) { e.printStackTrace(); } finally { //释放锁 rlock.unlock(); } return "end"; }}复制代码方法名为tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command)：
// 使用lua脚本加锁// getName()传入KEYS[1]，表示传入解锁的keyName，这里是 String lockKey = coupon:ticket" + couponCode;// internalLockLeaseTime传入ARGV[1]，表示锁的超时时间，默认是30秒// getLockName(threadId)传入ARGV[2]，表示锁的唯一标识线程id RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) { //当第一个线程请求进来时会直接执行这段逻辑 return this.evalWriteAsync(this.getRawName(), LongCodec.INSTANCE, command, //判断传入的Redis的key是否存在，即String lockKey = "coupon:ticket" + couponCode; "if (redis.call('exists', KEYS[1]) == 0) then " + //如果不存在那么就设置这个key为传入值、当前线程id 即参数ARGV[2]的值（即getLockName(threahId)，并且将线程id的value值设置为1） "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + //给这个key设置有效时间，有效时间即参数ARGV[1](即internalLockLeaseTime的值)的时间 "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + //当第二个线程进来，Redis中的key已经存在(锁已经存在)，那么直接进这段逻辑 //判断这个Redis key是否存在且当前的这个key是否是当前线程设置的 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + //如果是的话，那么就进入重入锁的逻辑，利用hincrby指令将第一个线程进来将线程id的value值设置为1再加1 //然后每次释放锁的时候就会减1，直到这个值为0，这把锁就释放了，这点与juc的可重锁类似 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getRawName()), new Object[]{unit.toMillis(leaseTime), this.getLockName(threadId)});}复制代码 本文通过单机环境本地锁->集群环境锁演变->分布式锁的延伸，深入浅出的介绍了不同方式实现分布式锁的问题和改进之处，通过不同的例子详细展开说明，并且分析了Redission分布式锁的实现原理和实现。
链接：https://juejin.cn/post/7133062101387444260
| 留言与评论（共有 0 条评论） “” |
