一、基于setnx实现的分布式锁问题

重入问题：获得锁的线程应能再次进入相同锁的代码块，可重入锁能防止死锁。例如在HashTable中，方法用synchronized修饰，若在一个方法内调用另一个方法，不可重入会导致死锁。而synchronized和Lock锁都是可重入的。
不可重试：目前的分布式锁只能尝试一次，合理的情况是线程在获得锁失败后应能再次尝试。
超时释放：加锁时增加过期时间可防止死锁，但如果卡顿时间超长，虽采用了 lua 表达式防止删锁时误删别人的锁，但毕竟没有锁住，存在安全隐患。
主从一致性：若 Redis 提供主从集群，向集群写数据时，主机异步同步数据给从机，若同步前主机宕机，会出现死锁问题。

二、Redission 快速入门

引入依赖：根据项目需求引入 Redisson 相关依赖。

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient(){
        // 配置
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.150.101:6379")
            .setPassword("123321");
        // 创建RedissonClient对象
        return Redisson.create(config);
    }
}

配置 Redisson 客户端：进行 Redisson 客户端的配置。

@Resource
private RedissionClient redissonClient;

@Test
void testRedisson() throws Exception{
    //获取锁(可重入)，指定锁的名称
    RLock lock = redissonClient.getLock("anyLock");
    //尝试获取锁，参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
    boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);
    //判断获取锁成功
    if(isLock){
        try{
            System.out.println("执行业务");          
        }finally{
            //释放锁
            lock.unlock();
        }
        
    }
}

使用 Redission 的分布式锁：在VoucherOrderServiceImpl中注入RedissonClient，以使用 Redisson 的分布式锁功能。

@Resource
private RedissonClient redissonClient;

@Override
public Result seckillVoucher(Long voucherId) {
        // 1.查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2.判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀尚未开始！");
        }
        // 3.判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀已经结束！");
        }
        // 4.判断库存是否充足
        if (voucher.getStock() < 1) {
            // 库存不足
            return Result.fail("库存不足！");
        }
        Long userId = UserHolder.getUser().getId();
        //创建锁对象 这个代码不用了，因为我们现在要使用分布式锁
        //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //获取锁对象
        boolean isLock = lock.tryLock();
       
		//加锁失败
        if (!isLock) {
            return Result.fail("不允许重复下单");
        }
        try {
            //获取代理对象(事务)
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            //释放锁
            lock.unlock();
        }
 }

三、Redission 可重入锁原理

在分布式锁中，Redission 采用 hash 结构存储锁。大 key 表示锁是否存在，小 key 表示当前锁被哪个线程持有。下面分析 lua 表达式的三个参数：

KEYS[1]：锁名称。
ARGV[1]：锁失效时间。
ARGV[2]：id + ":" + threadId，即锁的小 key。

执行过程如下：

redis.call('hset', KEYS[1], ARGV[2], 1)，往 Redis 中写入数据，形成 hash 结构，如Lock{id + ":" + threadId : 1}。

"if (redis.call('exists', KEYS[1]) == 0) then " +
                  "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]);"

若当前锁存在，第一个条件不满足，接着判断redis.call('hexists', KEYS[1], ARGV[2]) == 1，通过大 key 和小 key 判断当前锁是否属于自己。若是自己的，则执行redis.call('hincrby', KEYS[1], ARGV[2], 1)，将锁的 value 加 1，并执行redis.call('pexpire', KEYS[1], ARGV[1])设置过期时间。若以上两个条件都不满足，则抢锁失败，返回锁的失效时间。

查看源码会发现，会判断当前方法的返回值是否为null。若为null，对应前两个条件，退出抢锁逻辑；若返回值不是null，即走第三个分支，在源码处会进行while(true)的自旋抢锁。

四、Redission 锁重试和 WatchDog 机制

抢锁过程中，获得当前线程，通过tryAcquire进行抢锁，逻辑与之前相同：

先判断当前锁是否存在，若不存在，插入一把锁，返回null。
判断当前锁是否属于当前线程，若是，则返回null。

若返回值为null，代表当前线程已抢锁完毕或可重入完毕；若以上两个条件都不满足，则进入第三个条件，返回锁的失效时间。

long threadId = Thread.currentThread().getId();
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
    return;
}

接下来根据lock方法的重载情况进行处理。若传入参数，leaseTime不为-1，则进行抢锁；

if (leaseTime != -1) {
    return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
}

若没有传入时间，也会进行抢锁，且抢锁时间是默认看门狗时间。

commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()

ttlRemainingFuture.onComplete((ttlRemaining, e)

这句话相当于对抢锁进行监听，抢锁完毕后会调用特定方法开启一个线程进行续约逻辑，即看门狗线程。

RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                        commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                        TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
    if (e != null) {
        return;
    }

    // lock acquired
    if (ttlRemaining == null) {
        scheduleExpirationRenewal(threadId);
    }
});
return ttlRemainingFuture;

续约逻辑是通过commandExecutor.getConnectionManager().newTimeout()方法实现的，该方法表示在一定时间后执行特定任务。以锁失效时间为 30s，10s 后触发任务进行续约，将锁续约成 30s，若操作成功，会递归调用自己，重新设置任务，实现不停续约。若线程出现宕机，则不会续约，等到时间后自然释放锁。

五、Redission 锁的 MutiLock 原理

为提高 Redis 的可用性，通常会搭建集群或主从。以主从为例，写命令在主机上，主机会将数据同步给从机，但在主机还未将数据写入从机时宕机，哨兵会选举一个 slave 变成 master，此时新的 master 中没有锁信息，锁就丢失了。

Redission 提出 MutiLock 锁来解决这个问题。使用 MutiLock 锁不使用主从，每个节点地位相同，加锁逻辑需写入到每个节点上，只有所有服务器都写入成功才是加锁成功。若某个节点挂了，只要有一个节点拿不到锁，都不算加锁成功，保证了加锁的可靠性。

当设置多个锁时，Redission 会将多个锁添加到一个集合中，用while循环不停尝试拿锁，但有一个总共的加锁时间，为需要加锁的个数乘以 1500ms。例如有 3 个锁，时间就是 4500ms，在这时间内所有锁加锁成功才算加锁成功，若有线程加锁失败，则会再次重试。