一、案例

案例是一个简易版的电商项目中的下单功能，先简单分析下下单功能的一些基本步骤：
1.校验库存 2.保存订单 3.生成快照 4.扣减库存 5.删除购物车

二、问题

在这个案例中不难发现，在没有枷锁的情况下，如果两个线程同时执行下单操作，可能会在校验库存后同时执行后面的业务导致同时扣减库存，会引发库存小于0的情况，也就是超卖问题

在单体项目中，这个问题很容易解决，那就是加锁，使用一个 synchronized锁的可以很容易的解决

但是在分布式的项目中，你的后端服务器有多个，而 synchronized锁只能保证在同一个JVM中不会有并发问题，所以此时用 synchronized锁就不能解决问题了，此时就需要分布式锁

三、基于Redis实现分布式锁

1.实现

如上图，通过Redis的sexnx（key不在，添加；key在，不添加）来实现加锁，key设置为商品id，value暂时可以随便设置。在每次执行下单前，先使用sexnx来对商品加锁，如果对应的商品id在redis中不存在，那么该商品就是无锁的，可以加锁然后执行后序下单业务，如果对应商品id在redis中存在，那么说明该商品已经被加锁了，有其他线程正在操作这个商品，那么此时就需要等待或者直接下单失败了

代码实现：

 @Transactionalpublic String saveOrder(int productId) {    Boolean value = stringRedisTemplate.boundValueOps(productId + "").setIfAbsent("value");    if (value) {        try {            //检验库存            int stock = tbProductDao.querystock(productId);            if (stock > 0) {                //保存订单                System.out.println("保存订单");                //修改库存                tbProductDao.redustock(productId);                //删除购物车                return "订单提交成功";            } else {                return "商品不存在";            }        }catch (Exception e){            e.printStackTrace();        }finally {            //解锁            stringRedisTemplate.delete(productId+"");        }    }    return "订单提交失败";}

2. 优化

以上代码可以实现一个简单的分布式锁，但是仍然有一些可以优化的地方：
1.获取锁失败时会直接返回订单提交失败，应该不断尝试获取锁
2.无法保证锁的获取顺序是按照线程顺序来的

1）阻塞锁和非阻塞锁

对于问题1，可以通过阻塞锁或者非阻塞锁实现

阻塞锁：不断尝试获取锁，直到获取到锁为止

        //阻塞锁实现        while( b == false){            b = stringRedisTemplate.boundValueOps(productId + "").setIfAbsent("value");

非阻塞锁：如果获取不到锁就放弃，但是可以支持在一段时间内重试

 //非阻塞锁实现        int count = 1;        while( b == false && count < 4){            b = stringRedisTemplate.boundValueOps(productId + "").setIfAbsent("value");            count ++;        }

使用两种锁都可以实现
我们通过非阻塞锁来实现

2）公平锁和非公平锁

对于问题2 可以使用公平锁来实现
公平锁：按照线程的先后顺序获取锁
非公平锁：多个正在等待的线程随机获取锁

3.问题

1）线程异常导致无法释放锁

上述的分布式锁还有一个问题：
在执行业务的过程中，如果当前线程出现异常（宕机）没有释放锁，那么就会导致死锁

解决：
可以对锁设置过期时间，当出现异常没有释放锁，在过期时间结束时也能自动释放锁

2）T1过期释放T2锁

对锁设置过期时间能够解决死锁问题，但是另一个问题随之而来：假设有两个线程T1、T2，T1首先获得锁，但是在锁过期之前并没有完成业务的执行，同时T2线程成功加锁，T1执行结束后又释放锁（此时释放的是T2的锁），这就会导致T2在无锁状态下执行

解决：
在加锁时为锁设置唯一的value，释放锁时要先获取对应的value

 String value = UUID.randomUUID().toString(); boolean b = stringRedisTemplate.boundValueOps(productId + "")            .setIfAbsent(value,3,TimeUnit.MINUTES);

如果获取的指与当前value相同则释放锁

//查询操作String v = stringRedisTemplate.boundValueOps(productId + "").get();if(value.equals(v)){    //删除操作    stringRedisTemplate.delete(productId+"");}

简单来说就是每个线程只能释放自己加的锁

但是此时又产生了新的问题，看释放锁的代码，释放锁是分为两步进行的：1.查询value 2.删除
那么在这一步时也会产生并发问题，如果在查询完成后刚要删除锁，但是此时锁过期了并且其他线程成功加锁，那么也会导致本线程释放其他线程的锁

解决：
要解决这个问题本质是就是让查询和删除这两步操作是原子性的，可以使用lua脚本
lua：

if redis.call("get",KEYS[1]) == ARGV[1] then    return redis.call("del",KEYS[1])else    return 0end

配置Bean加载lua

@Beanpublic DefaultRedisScript<List> defaultRedisScript(){    DefaultRedisScript<List> defaultRedisScript = new DefaultRedisScript<>();    defaultRedisScript.setResultType(List.class);    defaultRedisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("unlock.lua")));    return  defaultRedisScript;}

执行lua解锁

@AutoWiredprivate DefaultRedisScript defaultRedisScript;//执行lua脚本List<String> keys = new ArrayList<>();keys.add(skuId);List rs = stringRedisTemplate.execute(defaultRedisScript,keys , values.get(skuId));System.out.println(rs.get(0));

分析以上的问题不难看出，对于过期时间的设置是一个很困难的问题，必须保证在这个时间内业务能够执行完成并且也不能等待太久
对于这个问题可以使用看门狗机制

4）看门狗机制

看门狗线程工作原理：
监听当前线程锁的过期时间，当锁即将过期时如果有任务没有执行结束，则重置锁的过期时间，保证有任务线程正常执行的过程中，锁不会过期

4.Redisson

基于以上的问题，我们可以使用Redisson------一个基于Redis+看门狗机制的分布式锁框架

依赖：

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

配置yml

redisson:  addr:    singleAddr:      host: redis://localhost:6380      database: 0

RedissonConfig：

@Configurationpublic class RedissonConfig {    @Value("${redisson.addr.singleAddr.host}")    private String host;    @Value("${redisson.addr.singleAddr.database}")    private int database;    @Bean    public RedissonClient redissonClient(){        Config config = new Config();        config                .useSingleServer()                .setAddress(host)                .setDatabase(database);        return Redisson.create(config);    }}

使用：

@Transactionalpublic String saveOrder(int productId) {        //获取公平锁        RLock lock=redissonClient.getFairLock(productId+"");        //非阻塞锁        try {            boolean b = lock.tryLock(3, TimeUnit.MINUTES);            if(b){                System.out.println("业务执行");                return "订单提交成功";            }        } catch (InterruptedException e) {            e.printStackTrace();        }finally {            lock.unlock();        }        return "订单提交失败";}