分布式锁

Synchronized弊端

Synchronized中的重量级锁，底层就是基于 锁监视器（Monitor） 来实现的。简单来说就是 锁对象头会指向一个锁监视器 ，而在监视器中则会记录一些信息，比如：

• _owner：持有锁的线程
• _recursions：锁重入次数

因此每一个锁对象，都会指向一个锁监视器，而每一个锁监视器，同一时刻只能被一个线程持有，这样就实现了互斥效果。但前提是，多个线程使用的是同一把锁。

比如有三个线程来争抢锁资源，线程1获取锁成功，如图所示：

此时其它线程想要获取锁，会发现监视器中的_owner已经有值了，就会获取锁失败。由于咱们代码在锁对象是用户id的字符串常量，因此同一个用户肯定是同一把锁，线程是绝对安全的。

但问题来了，我们的服务将来肯定会多实例不是，形成集群。每一个实例都会有一个自己的JVM运行环境，因此即便是同一个用户，如果并发的发起了多个请求，由于请求进入了多个JVM，就会出现多个锁对象（用户id对象），自然就有多个锁监视器。此时就会出现 每个JVM内部都有一个线程获取锁成功 的情况，没有达到互斥的效果，并发安全问题就可能再次发生了：

可见，在集群环境下，JVM提供的传统锁机制就不再安全了。

那么该如何解决这个问题呢？

显然，我们不能让每个实例去使用各自的JVM内部锁监视器，而是应该在多个实例外部寻找一个锁监视器，多个实例争抢同一把锁。

像这样的锁，就称为分布式锁。

分布式锁必须要满足的特征：

• 多JVM实例都可以访问
• 互斥

能满足上述特征的组件有很多，因此实现分布式锁的方式也非常多，例如：

• 基于MySQL
• 基于Redis
• 基于Zookeeper
• 基于ETCD

但目前使用最广泛的还应该是基于 Redis的分布式锁。

简单分布式锁

Redis本身可以被任意JVM实例访问，同时Redis中的setnx命令具备互斥性，因此符合分布式锁的需求。不过实现分布式锁的时候还有一些细节需要考虑，绝不仅仅是setnx这么简单。

基本原理

Redis的setnx命令是对string类型数据的操作，语法如下：

# 给key赋值为value
SETNX key value

当前 仅当key不存在的时候，setnx才能执行成功，并且返回1，其它情况都会执行失败，并且返回0 .我们就可以认为 返回值是1就是获取锁成功，返回值是0就是获取锁失败 ，实现互斥效果。 而当业务执行完成时，我们只需要删除这个key即可释放锁。这个时候其它线程又可以再次获取锁（执行setnx成功）了。

# 删除指定key，用来释放锁
DEL key

例如，我们用lock作为某个业务的锁的key，获取锁就执行下面命令：

# 获取锁，并记录持有锁的线程
SETNX lock thread1

假设说一开始lock不存在，有很多线程同时对lock执行setnx命令。由于Redis命令本身是串行执行的，也就是各个线程是串行依次执行。因此当第一个线程执行setnx时，会成功添加这个lock。但其余的线程会发现lock已经存在，自然就执行失败。自然就实现了 互斥效果。 当业务执行完毕，直接删除lock，自然就释放锁了：

# 释放锁
DEL lock

不过我们要考虑一种极端情况，比如我们获取锁成功，还未释放锁呢当前实例突然宕机了 ！那么释放锁的逻辑自然就永远不会被执行，这样 lock就永远存在，再也不会有其它线程获取锁成功了！出现了死锁问题。 怎么办？

我们可以利用Redis的KEY过期时间机制，在获取锁时给锁添加一个超时时间：

# 获取锁，并记录持有锁的线程
SETNX lock thread1
# 设置过期时间，避免死锁
EXPIRE lock 20

这里我们设置超时时间为20秒，远超任务执行时间。当业务正常执行时，这个过期时间不起作用，我们通过DEL命令来释放锁。 但是如果当前服务实例宕机，DEL无法执行。但由于我们 设置了20秒的过期时间，当超过这个时间时，锁会因为过期被删除 ，因此就等于释放锁了，从而避免了死锁问题。这种策略就是超时释放锁策略。

但新的问题来了，SETNX和EXPIRE是两条命令，如果我执行完SETNX，还没来得急执行EXPIRE时服务已经宕机了，这样加锁成功，但锁超时时间依然没能设置！死锁问题岂不是再次发生了？！

所以，为了解决这个问题，我们必须保证SETNX和EXPIRE两个操作的原子性。事实上，Redis中的set命令就能同时实现setnx和expire的效果：

# NX 等同于SETNX lock thread1效果；
# EX 20 等同于 EXPIRE lock 20效果
SET lock thread1 NX EX 20

综上，利用Redis实现的简单分布式锁流程如下：

代码实现

@RequiredArgsConstructor
public class RedisLock {

    private final String key;
    private final StringRedisTemplate redisTemplate;

    /**
     * 尝试获取锁
     * @param leaseTime 锁自动释放时间
     * @param unit 时间单位
     * @return 是否获取成功，true:获取锁成功;false:获取锁失败
     */
    public boolean tryLock(long leaseTime, TimeUnit unit){
        // 1.获取线程名称
        String value = Thread.currentThread().getName();
        // 2.获取锁
        Boolean success = redisTemplate.opsForValue().setIfAbsent(key, value, leaseTime, unit);
        // 3.返回结果
        return BooleanUtils.isTrue(success);
    }

    /**
     * 释放锁
     */
    public void unlock(){
        redisTemplate.delete(key);
    }
}

分布式锁的问题

锁误删问题

第一个问题就是锁误删问题，目前释放锁的操作是基于DEL，但是在极端情况下会出现问题。

例如，有线程1获取锁成功，并且执行完任务，正准备释放锁：

但是因为某种原因导致释放锁的操作被阻塞了，直到锁被超时释放：

就在此时，有一个新的线程2来尝试获取锁。因为线程1的锁被超时释放，因此线程2是可以获取锁成功的：

而就在此时，线程1醒来，继续执行释放锁的操作，也就是DEL.结果就把线程2的锁给删除了：

然而此时线程2还在执行任务，如果有其它线程再来获取锁，就会认为无人持有锁从而获取锁成功，于是多个线程再次并行执行，并发安全问题就可能再次发生了：

解决思路：

还记得我们set时存入了什么吗？

SET lock thread1 NX EX 10

我们会将持有锁的线程存入lock中。因此，我们应该在 删除锁之前判断当前锁的中保存的是否是当前线程标示，如果不是则证明不是自己的锁，则不删除；如果锁标示是当前线程，则可以删除：

综上，分布式锁的实现逻辑就变化了：

超时释放问题

加上了锁标示判断逻辑，可以避免大多数情况下的锁误删问题，但是还有一种极端情况依然会存在误删可能。

例如，线程1获取锁成功，并且执行业务完成，并且也判断了锁标示，确实与自己一致：

接下来，线程1应该去释放自己的锁了，可就在此时发生了阻塞！直到锁超时释放：

此时，如果有线程2来获取锁，肯定可以获取锁成功：

就在线程2获取锁成功后，线程1从阻塞中醒来，继续释放锁。由于在 阻塞之前已经完成了锁标示判断，现在就无需判断而是直接删除锁，结果就把线程2的锁删除了：

有一次发生了误删问题！！尴尬不 总结一下，误删的原因归根结底是因为什么？

• 超时释放
• 判断锁标示、删除锁两个动作不是原子操作

超时释放不能不做，因为要避免服务宕机导致的死锁，必须加超时时间。但是加了超时时间又出现了误删问题。怎么办？ 操作锁的多行命令又该如何确保原子性？

其它问题 除了上述问题以外，分布式锁还会碰到一些其它问题：

• 锁的重入问题：同一个线程多次获取锁的场景，目前不支持，可能会导致死锁
• 锁失败的重试问题：获取锁失败后要不要重试？目前是直接失败，不支持重试
• Redis主从的一致性问题：由于主从同步存在延迟，当线程在主节点获取锁后，从节点可能未同步锁信息。如果此时主宕机，会出现锁失效情况。此时会有其它线程也获取锁成功。从而出现并发安全问题。
• ...

当然，上述问题并非无法解决，只不过会比较麻烦。例如：

• 原子性问题：可以利用Redis的LUA脚本来编写锁操作，确保原子性
• 超时问题：利用 WatchDog（看门狗）机制 ，获取锁成功时开启一个定时任务，在锁到期前自动续期，避免超时释放。而当服务宕机后，WatchDog跟着停止运行，不会导致死锁。
• 锁重入问题：可以模拟Synchronized原理，放弃setnx，而是利用 Redis的Hash结构来记录锁的持有者以及重入次数，获取锁时重入次数+1，释放锁是重入次数-1，次数为0则锁删除
• 主从一致性问题：可以利用Redis官网推荐的 RedLock 机制来解决

这些解决方案实现起来比较复杂，因此我们通常会使用一些开源框架来实现分布式锁，而不是自己来编码实现。目前对这些解决方案实现的比较完善的一个第三方组件：Redisson

因此，我们只要会使用Redisson，即可解决上述问题，无需自己动手编码了。

Redisson

Redisson是一个基于Redis的工具包，功能非常强大。将JDK中很多常见的队列、锁、对象都基于Redis实现了对应的分布式版本。 例如：

快速入门

<!--redisson-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
</dependency>

配置

 @Configuration
 public class RedisConfig {
    @Bean
    public RedissonClient redissonClient() {
        // 配置类
        Config config = new Config();
        // 添加redis地址，这里添加了单点的地址，也可以使用config.useClusterServers()添加集群地址 
        config.useSingleServer()
            .setAddress("redis://192.168.150.101:6379")
            .setPassowrd("123321");
        // 创建客户端
        return Redisson.create(config);
    }
 }

基本用法

@Autowired
 private RedissonClient redissonClient;

 @Test
 void testRedisson() throws InterruptedException {
    // 1.获取锁对象，指定锁名称
    RLock lock = redissonClient.getLock("anyLock");
    try {
        // 2.尝试获取锁，参数：waitTime、leaseTime、时间单位
        boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
        if (!isLock) {
            // 获取锁失败处理 ..
        } else {
            // 获取锁成功处理
        }
    } finally {
        // 4.释放锁
        lock.unlock();
    }
 }

利用Redisson获取锁时可以传3个参数：

• waitTime：获取锁的等待时间。当获取锁失败后可以多次重试，直到waitTime时间耗尽。waitTime默认-1，即失败后立刻返回，不重试。
• leaseTime：锁超时释放时间。默认是30，同时会利用WatchDog来不断更新超时时间。需要注意的是，如果手动设置leaseTime值，会导致WatchDog失效。
• TimeUnit：时间单位

总结

分布式锁用Redisson就行了，它以经为你解决的大部分问题