基于Redis的分布式锁

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

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

问题分析 

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

分布式锁有如下特点：

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

Redis锁的demo

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

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

分布式锁初步实现

实现锁接口

public class SimpleRedisLock implements ILock {

    //用户的userid
    private 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) + "-";

    @Override
    public 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，则返回false
        return Boolean.TRUE.equals(success);
    }

    @Override
    public 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 {

    //用户的userid
    private 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)+"-";

    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标示  线程前缀+线程id
        String threadId = THREAD_PREFIX + Thread.currentThread().getId();
        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);

         //防止Boolean和boolean拆箱出问题，如果success为null，则返回false
        return Boolean.TRUE.equals(success);
    }

    @Override
    public 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 {

    //用户的userid
    private 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);
    }
    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取线程标示  线程前缀+线程id
        String threadId = THREAD_PREFIX + Thread.currentThread().getId();
        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);

         //防止Boolean和boolean拆箱出问题，如果success为null，则返回false
        return Boolean.TRUE.equals(success);
    }
    @Override
    public 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]) then
	return redis.call('del',KEYS[1])
end
return 0