redis实现分布式锁

一，redis除了缓存，还可以用来干啥

分布式锁，分布式限流，session共享，延迟队列等

二，分布式场景

1，互联网秒杀
2，抢优惠券
3，接口幂等性校验

三，解决超卖问题

1，悲观锁，通过 for update 实现提前加锁，在锁住这条记录之后，别人则无法操作当前这行记录。不过在高并发的场景下，容易出现死锁的问题，会大大降低的mysql的性能
2，乐观锁，通过添加一个version版本实现
因此不可能在mysql上面进行操作，一般的话都会在redis上面进行操作，通过redis进行一个库存的预扣减。一个减库存的伪代码如下，通过redis实现库存的预扣减。

//获取库存
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if (stock > 0) {
	//每人限购一个
    int realStock = stock - 1;
    //更新库存
    stringRedisTemplate.opsForValue().set("stock", realStock + ""); 
    System.out.println("扣减成功，剩余库存:" + realStock + "");
} else {
    System.out.println("扣减失败，库存不足");
}

如果是部署在一个分布式架构或者集群上面，上面的代码就会出现问题，即会在多个tomcat下。如果只是简单的设置synchronized 的同步代码块，只能保证一个jvm下面的线程同步，即只能保证一个tomcat 下面的线程安全。因此要保证分布式场景下面线程安全，因此需要使用到分布式锁。分布式锁主要有mysql，redis，zookeeper，这里主要分析一下redis的分布式锁的使用场景以及原理流程

setnx key value
他的底层是一个原子操作， 它可以在同一时间内完成设置值和设置过期时间这两个操作， 因此 SETEX 命令在储存缓存的时候非常实用。 因此可以解决上述的分布式锁的问题可以如下

//设置锁
String lockKey = "product_001";
#获取锁，并且设置超时时间，如果业务出现异常，可以在指定时间内释放锁
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "hat", 30, TimeUnit.SECONDS);
#释放锁-
stringRedisTemplate.delete(lockKey);

由于多个线程都去抢同一把锁，所以在高并发的场景下，假设第一个线程获取锁lockKey，设置了过期时间为30s，但是如果业务逻辑需要的时间大于30s，则redis会因为这个过期时间释放锁，那么第二个线程就会获取锁成功，假设业务逻辑需要45s，那么还需要当前线程还需要15s将任务全部执行完毕，但是第二个线程获取锁后进来了，假设也需要45s，那么在第二个执行15s时，第一个线程全部执行完，就会进行锁的释放删除，但是全部线程用的是同一把锁，那么此时删除的锁就是把第二个进来的线程的锁给删除了，那么第三个线程又会进来，一直循环下去，将别的线程的锁给误删。因此有可能一直造成锁失效的问题，如果在高并发的场景下，很难保证线程的执行顺序，因此有可能会出现很大的线程安全问题。

通过上述分析，也就是说所有人都共用同一把锁，因此可以在释放锁的时候进行改进，为每个线程给予一个唯一的id进行客户端的标识，然后在进行锁删除的时候，需要通过这个唯一id进行锁的删除释放，在删除锁时需要进行这个客户端id的判断，在不出现异常的情况下，当前线程可以走完全部的业务逻辑，解决了上面锁被误删的情况，伪代码如下：

//将这个uuid作为客户端id，实现客户端的唯一标识
String clientId = UUID.randomUUID().toString();
//value则设置为这个uuid
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
//锁删除时进行判断，是否删除当前线程的锁
if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
    //释放锁，如果在这行代码出现宕机，也会死锁几秒
    stringRedisTemplate.delete(lockKey);
}

在释放锁的时候，也可能出现宕机的问题或者网络延迟的问题，假设代码执行到删除锁的那行，出现服务器卡的界面，那么也可能在删除这个锁的时候会将下一个线程的锁给删除，虽然概率比较小，因此也可能出现并发安全的问题。因此需要使用接下来的redisson的方式实现，通过锁续命的方式实现这个分布式锁的安全问题。

四，redisson

4.1，需要的依赖包如下

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

4.2，初始化redisson

@Bean
    public Redisson redisson() {
        // 此为单机模式
        Config config = new Config();
        config.useSingleServer().setAddress("redis://124.222.199.186:6379").setDatabase(0);
        return (Redisson) Redisson.create(config);
    }

4.3，redisson使用的基本伪代码如下

//设置一个key作为锁的标志，如果key存在则表示资源被锁
String lockKey = "product_001";
//获取redis的分布式锁
RLock redissonLock = redisson.getLock(lockKey);
// 加锁，实现锁续命功能,锁时间大概在30s左右，该锁也是一个重入锁
//因此加了多少个锁，就需要解多少个锁
redissonLock.lock();
redissonLock.lock();
//解锁
redissonLock.unlock();
redissonLock.unlock();

4.4，lock方法的底层实现以及流程

//如果锁不可用，那么当前线程将被禁用，并处于休眠状态，直到获得锁。
void lock();

1，在多个线程的情况下，会有其中的一个线程获取到锁，获取锁的线程会进行加锁操作，通过主线程执行，分线程续命方式，会后台开启一个守护进程，该线程被称为watch dog，就是看门狗的意思，每隔10s检查当前线程的锁是否过期，当前线程还是否持有锁，如果持有锁的话则延长锁的时间，最后释放锁
2，如果获取锁的进程出现宕机的情况，并不会出现死锁的情况，会设置一个30s的过期时间，30s后自动释放锁
3，当持有锁的主线程宕机之后，主线程对应的守护线程也会随着被销毁，锁也就会自动释放
4，其他线程也会通过while循环，即自旋的方式获取锁，直到加锁成功
5，redis服务端会记录当前是哪个线程进行了加锁，有与时重入锁，因此会通过一个计数器来记录加锁的次数，因此在解锁的时候，加了多少次 lock 锁，就需要进行多少次 unlock 的解锁
6，redis使用了大量的原子操作，但是redis操作只保证自己的原子性，因此运用了大量的原子操作，通过lua脚本来实现redis批量操作的原子性。

4.5，redis Lua脚本

1，redis Lua脚本的优点
1，减少网络开销，即通过一次请求完成多次命令的操作
2，原子操作，可以让redis的批量操作实现原子性
3，替代redis的事务功能，支持常规事务，以及回滚操作
2，举例

//初始化商品10016的库存
jedis.set("product_stock_10016", "15");  
				//定义一个本地变量count，redis.call函数实现redis命令
String script = " local count = redis.call('get', KEYS[1]) " + 
    			//转成数字类型
				" local a = tonumber(count) " +
    			//ARGV[1] 对应的value
                " local b = tonumber(ARGV[1]) " +
    			//判断如果a>b,则扣减库存，否则语法报错
                " if a >= b then " +
                " redis.call('set', KEYS[1], count-b) " +
                //模拟语法报错回滚操作"   bb == 0 " +
                "   return 1 " +
                " end " +
                " return 0 ";
//执行这段脚本
Object obj = jedis.eval(script, Arrays.asList("product_stock_10016"), Arrays.asList("10"));
System.out.println(obj);
#因为redis是单线程，因此如果lua脚本尽量不要写的太复杂，否则可能造成后面的任务不执行

redisson 底层通过lua脚本实现redis分布式锁的原理如下。

return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                       #hash存储对象
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

五，最终优化后的代码如下

private Redisson redisson;
private StringRedisTemplate stringRedisTemplate;
public String deductStock() throws InterruptedException {
        //获取锁
        String lockKey = "product_001";
    	//给每一个线程一个唯一id标识
        //String clientId = UUID.randomUUID().toString();
        RLock redissonLock = redisson.getLock(lockKey);
        try {
            //原生的方式获取锁  jedis.setnx(key,value)
            //Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "zhenghuisheng");
            //设置超时时间
            //stringRedisTemplate.expire(lockKey,30, TimeUnit.SECONDS);
            //上面两条命令的组合
            /*Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);

            if (!result) {
                return "error_code";
            }*/

            // 加锁，实现锁续命功能
            redissonLock.lock();
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if (stock > 0) {
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + "");
                System.out.println("扣减成功，剩余库存:" + realStock + "");
            } else {
                System.out.println("扣减失败，库存不足");
            }
        }finally {
            redissonLock.unlock();
            /*if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
                    //释放锁，如果在这行代码出现宕机，也会死锁几秒
                    stringRedisTemplate.delete(lockKey);
            }*/
        }
        return "end";
    }

这样的话，就能保证在高并发场景下解决线程安全问题和超卖问题了。