基于Redis的分布式锁

分布式场景下并发安全问题的引发

前面通过加锁解决了单机状态下一人一单的问题，但是当出现了分布式，前面的加锁形式不再适用 ，每个jvm有一个自己的锁监视器，只能被内部线程获取，其他jvm无法使用，那么多台jvm的锁监视器不共用一个锁监视器，就容易出现分布式场景下并发安全问题。

问题分析 

所以我们要使用可以解决分布式场景下的位于jvm外的锁，多个jvm共同使用该锁，而不是使用每个jvm的内部锁。

分布式锁有如下特点：

 这里我们就选用redis来实现我们的分布式锁！

Redis锁的demo

redis锁要实现如上两个基本操作：获取锁和删除锁，在获取锁的同时为了防止宕机出现死锁，要手动添加过期时间，那么为了防止只加锁没有加过期时间的情况出现，我们要保证加锁和加过期时间的原子性，也就是他俩必须同时进行！

那么上述加锁的命令可以换成如下：

分布式锁初步实现

实现锁接口

public class SimpleRedisLock implements ILock {//用户的useridprivate String name;private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}//锁的key值private static final String KEY_PREFIX = "lock:";//生成锁的value值private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";@Overridepublic boolean tryLock(long timeoutSec) {// 获取线程标示String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);//防止Boolean和boolean拆箱出问题，如果success为null，则返回falsereturn Boolean.TRUE.equals(success);}@Overridepublic void unlock() {//释放锁stringRedisTemplate.delete(KEY_PREFIX + name);}
}

使用锁

在VoucherOrderServiceImpl.java中的seckillVoucher方法中编写如下代码：

//获取用户userid
Long id = UserHolder.getUser().getId();SimpleRedisLock redisLock = new SimpleRedisLock("order:" + id, stringRedisTemplate);
//加锁，1200s是锁的过期时间boolean tryLock = redisLock.tryLock(1200);//判断锁是否获取成功if (!tryLock){return Result.fail("不允许重复下单");}try {//锁加到这里，事务提交后才释放锁//获取事务的动态代理对象，需要在启动类加注解暴漏出对象IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();//拿到动态代理对象
//            return createVoucherOrder(voucherId, voucher);return proxy.createVoucherOrder(voucherId, voucher);//使用动态代理类的对象，事务可以生效} finally {//无论如何都要释放锁，防止死锁redisLock.unlock();}

分布式场景下调试看效果

两个application，一个是8081端口，一个是8082端口。

apifox模拟同一个用户发送请求，authorization的参数值是同一个用户的，存储在redis中。

 如下：在8082的断点处获取锁失败，在8081的断点处获取锁成功，即只有一次成功获取锁。

数据库中优惠券库存stock-1而不是-2，优惠券订单产生1个，数据库没问题！

redis中查看锁的key值，1010正是userid，问题解决，达到我们想要的效果！

redis分布式锁误删问题

问题分析

当线程1获取锁成功时，如果该业务执行时间长以至于超过了设置的锁过期时间，那么在业务还未完成时，锁便自动释放，此时线程1无锁，线程2获取到了锁执行业务，当线程1业务执行完后，按照业务逻辑仍会释放锁，但此时释放的是线程2的锁，这就出现了锁误删的问题。

解决锁误删

对于每个线程，我们获取其线程标识（每个JVM内部都维护了线程的id，这个id是自增的，那么多个jvm可能出现线程id一致的情况，为了避免该情况出现我们用UUID生成一个随机字符串作为前缀，以降低线程id重复的概率）作为锁的value

在释放锁时，我们先从redis获取对应的value值，跟当前线程的value做对比，一致则可以删除，否则就不能删除。

修改trylock和unlock方法 

public class SimpleRedisLock implements ILock {//用户的useridprivate String name;private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}//锁的key值private static final String KEY_PREFIX = "lock:";//线程的前缀，因为分布式下，多个jvm，每个jvm中维护的线程的id都是递增的，那么可能出现多个jvm的线程id一致，所以这里用uuid生成字符串作为前缀private static final String THREAD_PREFIX = UUID.randomUUID().toString(true)+"-";@Overridepublic boolean tryLock(long timeoutSec) {// 获取线程标示  线程前缀+线程idString threadId = THREAD_PREFIX + Thread.currentThread().getId();// 获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);//防止Boolean和boolean拆箱出问题，如果success为null，则返回falsereturn Boolean.TRUE.equals(success);}@Overridepublic void unlock() {// 获取线程标示String threadId =THREAD_PREFIX + Thread.currentThread().getId();// 获取锁中的标示String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);// 判断锁标示是否一致，防止锁误删if(threadId.equals(id)) {// 释放锁stringRedisTemplate.delete(KEY_PREFIX + name);}}
}

分布式锁的原子性

问题分析

JVM做Full GC时会阻塞所有代码，时间过长会出现锁超时自动释放，那么其他线程会趁虚而入获得锁。 

那么会出现如下情况：线程1获取锁执行业务逻辑后要释放锁，在判断完释放锁的条件为true后，即 threadId.equals(id)==true，正要释放锁时出现 Full GC，所有代码被阻塞，直到锁超时自动释放 （注意此时锁不是正常释放而是锁超时释放的），就在这时GC完毕代码恢复，线程2趁虚而入获得锁，而线程1也恢复了要执行释放锁的代码，因为GC前已经判断过释放条件为ture，那么此时线程1仍然认为锁是自己的，会错误地释放线程2的锁，又出现了误删问题。这里我们就要保证锁的原子性，即 判断锁的标识 和 释放锁 两个动作必须同时发生！

问题解决（Lua脚本）

Lua是一种编程语言，Redis提供了Lua脚本功能，即可以在一个脚本中写多条redis指令，确保了多条命令执行的原子性

代码实现

public class SimpleRedisLock implements ILock {//用户的useridprivate String name;private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}//锁的key值private static final String KEY_PREFIX = "lock:";//线程的前缀，因为分布式下，多个jvm，每个jvm中维护的线程的id都是递增的，那么可能出现多个jvm的线程id一致，所以这里用uuid生成字符串作为前缀private static final String THREAD_PREFIX = UUID.randomUUID().toString(true)+"-";private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}@Overridepublic boolean tryLock(long timeoutSec) {// 获取线程标示  线程前缀+线程idString threadId = THREAD_PREFIX + Thread.currentThread().getId();// 获取锁Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);//防止Boolean和boolean拆箱出问题，如果success为null，则返回falsereturn Boolean.TRUE.equals(success);}@Overridepublic void unlock() {// 调用lua脚本  stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),THREAD_PREFIX + Thread.currentThread().getId());}
}

在resource资源文件夹下创建Lua脚本内容如下：

if (redis.call('get',KEYS[1])==ARGV[1]) thenreturn redis.call('del',KEYS[1])
end
return 0