Redis实现分布式锁

1. 为什么需要分布式锁？?

我们都知道，Java提供的synchronized / lock锁只能作用在单个JVM中的单个应用中，Java中的锁只能锁定JVM级别的锁：

• synchronized就是利用JVM内部的锁监视器来控制线程的，在JVM的内部，因为只有一个锁监视器，所以只能有一个线程获取到锁，实现线程间的互斥；但是当有多个JVM的时候，就会有多个锁监视器，此时就会有多个线程获取到锁，这个时候就没有办法实现多JVM进程之间的互斥了。

而如果在以下场景下，比如：

• 单个应用进行集群部署，负载均衡可能把请求分配到不同的机器上
• 多个不同的分布式应用，多个应用需要同时锁定同一个资源

以上情况，单机锁就会失效，此时就需要一种全局应用锁代替单机锁，即：分布式锁。?

2. 单机锁 & 分布式锁的定义：?

• 单机锁：在单个JVM进程内起作用的同步机制，用于控制对共享资源的访问，这种锁，主要适用于单体应用，用于控制在同一JVM进程中多个线程对共享资源的互斥访问。
• 分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁，分布式锁用于协调分布式系统中的不同节点，以确保在全局范围内对共享资源的互斥访问。

锁的本质：让程序并行执行变成串行执行~！

3. 分布式锁应该具备哪些条件？

• 多线程可见
• 互斥
• 可重入：同一个线程可以反复获取锁，避免死锁
• 具备非阻塞锁特性：获取锁失败立即返回
• 获取锁和释放锁要具备高可用、高性能?

4. 分布式锁都有哪些主流的实现方案？

分布式锁的核心是实现多线程之间互斥，而满足这一点的方式有很多，常见的有三种：?

• Redis：简单，速度快，性能最好，现在企业级开发基本都使用Redis或者Zookeeper作为分布式锁，简单的SET NX EX这样的互斥命令就可以实现，如果插入key成功，则表示获取到了锁，如果插入失败则表示获取锁失败，而且还可以自动过期，利用锁超时时间自动释放，防止忘记释放锁
• MySQL：MySQL本身就带有锁机制（select...from...where..for update => 这是一种悲观锁，会锁住对应的索引行），利用MySQL本身的互斥锁机制来实现分布式锁，不用引入新的中间件，而是作为传统关系型数据库，存储的锁信息更详细，断开连接时自动释放锁，但是由于MySQL性能本身一般（锁的性能受限于MySQL的性能），所以使用MySQL作为分布式锁比较少见
• Zookeeper：Zookeeper也是企业级开发中较好的一个实现分布式锁的方案，利用临时顺序节点实现互斥锁，利用节点的唯一性和有序性实现互斥，断开连接时自动释放锁，更稳定更可靠，它不依靠超时时间释放锁，可靠性比Redis更高，同时也更复杂

5. 如何用 Redis 实现分布式锁的？?

Redis本身可以被多个客户端访问，正好就是一个共享存储系统，可以用来保存分布式锁，而且Redis的读写性能高，可以应对高并发的锁操作场景。

利用Redis的单线程特性，在多个Redis客户端同时通过SETNX命令尝试获取锁时，可以保证只会有一个客户端获取到锁，而其它客户端则会获取锁失败。

Redis的SET命令有个NX参数可以实现「key不存在才插入」，所以可以用它来实现分布式锁：?

• 如果key不存在，则显示插入成功，可以用来表示加锁成功；
• 如果key存在，则会显示插入失败，可以用来表示加锁失败。?

del或UNLINK删除key代表释放锁?

另外，我们可以在SET命令执行时加上 EX / PX 选项，设置过期时间，以免客户端拿到锁后发生异常，导致锁一直无法释放，防止死锁。?

实现分布式锁时需要实现的两个基本方法：

• 获取锁：

  • 互斥：确保只能有一个线程获取到锁? =>? 添加锁，利用setnx的互斥特性

  • 非阻塞：尝试一次，成功则返回true，失败则返回false

• 释放锁：

  • 手动释放：DEL? KEY手动删除

  • 超时释放：获取锁时添加一个超时时间：EX（推荐使用，因为一次就设置了 => 保证加锁和增加过期时间具有原子性）? 或? EXPRIE（需要设置两次，两次命令之间有可能发生异常）

6. 基于Redis实现分布式锁

• 加锁逻辑

锁的基本接口

package com.gch.redis.lock;

/**
 * 锁的基本接口
 */
public interface ILock {
    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间,过期后自动释放
     * @return true代表获取锁成功;false代表获取锁失败
     */
    boolean tryLock(long timeoutSec);

    /**
     * 释放锁
     */
    void unlock();
}

package com.gch.redis.lock;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 利用Redis实现分布式锁
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Component
public class SimpleRedisLock implements ILock {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private static final String KEY_PREFIX = "lock:";
    private String keyName;

    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间,过期后自动释放
     * @return true代表获取锁成功;false代表获取锁失败
     */
    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取当前的线程ID
        Long tid = Thread.currentThread().getId();
        // 获取锁 Boolean.TRUE.equals():防止空指针异常
        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + keyName, tid + "", timeoutSec, TimeUnit.SECONDS));
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        // 通过del删除锁
        stringRedisTemplate.delete(KEY_PREFIX + keyName);
    }
}

7. Redis分布式锁误删情况说明

逻辑说明：

• 当第一个持有锁的线程在锁的内部出现了阻塞，导致它的锁自动释放（兜底方案 => 超时释放锁），这时候线程2来尝试获得锁成功，然后线程2持有锁在执行的过程当中，线程1被唤醒来继续执行，而线程1在执行的过程中，走了删除锁或释放锁的逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明。?

解决Redis分布式锁的误删问题

在获取锁时，存入自己线程的标识，在删除释放锁时先获取锁中的线程标识，判断是否与当前线程标识一致，判断当前这把锁的标识是不是自己存入的：

• 如果一致则释放锁
• 如果不一致则不释放锁

思考：如果是在集群的环境下，线程的标识用线程ID表示还合适吗？？

• 当然不合适了，多JVM进程下的线程ID肯定有可能会冲突，导致两个不同的线程ID出现重复，因为单JVM进程中的线程ID它是依次递增的。
• ?因此可以用UUID来去表示? ?=>? 利用UUID + 线程ID。

package com.gch.redis.lock;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 利用Redis实现分布式锁
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Component
public class SimpleRedisLock implements ILock {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private String keyName;
    private static final String KEY_PREFIX = "lock:";
    private static final String UUID = java.util.UUID.randomUUID() + " - ";

    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间,过期后自动释放
     * @return true代表获取锁成功;false代表获取锁失败
     */
    @Override
    public boolean tryLock(long timeoutSec) {
        // 获取当前的线程ID
        String tid = UUID + Thread.currentThread().getId();
        // 获取锁 Boolean.TRUE.equals():防止空指针异常
        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + keyName, tid, timeoutSec, TimeUnit.SECONDS));
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        // 获取当前线程标识
        String tid = UUID + Thread.currentThread().getId();
        // 获取锁中的线程标识
        String kid = stringRedisTemplate.opsForValue().get(KEY_PREFIX + keyName);
        // 判断标识是否一致
        if(kid.equals(tid)){
            // 通过del删除锁
            stringRedisTemplate.delete(KEY_PREFIX + keyName);
            System.out.println("释放锁成功...");
        }
        System.out.println("释放锁失败...");
    }
}

8.?分布式锁的原子性问题

更为极端的误删逻辑说明：

• 线程1在持有锁完成业务逻辑之后，正准备删除锁，而且已经走到了条件判断的过程中，比如它已经通过判断当前这把锁确实是它自己的，正准备执行删除锁的逻辑时，此时发生了阻塞，导致删除锁失败，比如JVM发生Full GC就会导致所有业务代码阻塞，一旦阻塞时间过长导致锁超时被自动释放，而此时GC也已经结束，此时就会有其它线程来获取锁成功，比如线程2，此时线程1也不再阻塞而是被唤醒，由于刚才已经判断通过了，所以此时它又去接着执行删除锁的逻辑，它认为锁还是自己的，但其实这个时候锁已经被别的线程持有了，于是就把线程2的锁给释放掉了，又一次发生了误删。
• 这就是删除时的原子性问题，是因为线程1的获取锁、判断锁标识、以及删除释放锁这三个过程实际上并不是原子性操作，所以我们要防止刚才的情况发生。?

Redis的事务能够保证原子性，但无法保证事务的一致性，而且Redis的事务里面的多个操作其实是一个批处理，是在最终一次性去执行的，所以也就导致没有办法把获取锁、判断锁标识、以及删除释放锁这三个动作放在一个事务当中去执行，因此不推荐使用Redis的事务? =>? 推荐使用Lua脚本解决多条命令的原子性问题。

Redis的Lua脚本

• Redis提供了Lua脚本功能，Lua脚本其实就是在一个脚本当中编写多条Redis命令，把这多条命令放到一个脚本当中，并把它当作一个任务加入到一个队列当中作为一个整体进行执行，然后Redis单线程按照队列的顺序去依次执行这些任务，即Redis在执行的时候就会一次性去执行它们，在执行过程中Lua脚本不会被其它命令或请求打断，因此可以确保多条命令执行时的原子性。
• Lua脚本能够保证原子性是因为它在执行原子操作时会将其它线程或进程阻塞，直到该操作完成。

• Redis是C语言写的，任何C写的程序都可以穿插Lua脚本
• 之所以叫Lua脚本，是因为Lua是一种编程语言，它的基本语法可以参考网站：Lua 教程 | 菜鸟教程 ?

• 只需要了解Lua脚本有什么作用即可。。。?

这里重点介绍Redis提供的调用函数，语法如下：

redis.call('命令名称', 'key', '其它参数', ...)

?例如，我们要执行set name jack，则脚本是这样：

# 执行 set name jack
redis.call('set', 'name', 'jack')

例如，我们要先执行set name Rose，再执行get name，则脚本如下：

# 先执行 set name jack
redis.call('set', 'name', 'Rose')
# 再执行 get name
local name = redis.call('get', 'name')
# 返回
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

• 脚本的本质就是一个字符串，外面用双引号，里面就要用单引号，避免冲突
• 用EVAL函数来执行Lua脚本，保证解锁时的原子性

?例如，我们要执行 redis.call('set', 'name', 'jack') 这个脚本，语法如下：

如果脚本中的key、value不想写死，可以作为参数传递，key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

写死的就是常量 - 即内容是固定的，不写死的意思就是将来允许可以传参的，内容是不固定的。?

• 注意：在Lua语言里，数组的角标是从1开始的！?

9. 利用Java代码调用Lua脚本改造分布式锁

• Lua脚本本身并不需要大家花费太多时间去研究，只需要知道如何调用，大致是什么意思即可。?

需求：基于Lua脚本实现分布式锁的释放锁逻辑? ? =>? ?保证操作的原子性

提示：RedisTemplate调用Lua脚本的API如下?

我们的RedisTemplate中，可以利用execute方法去执行lua脚本，参数对应关系就如下图股：

 private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
    // 提前加载好Lua脚本
    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

    /**
     * 释放锁
     */
    public void unlock() {
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    }

• "泛型是给编译器看的，实际运行起不了多大的作用..."?

小总结：

基于Redis的分布式锁实现思路：

• 利用set nx ex获取锁，并设置过期时间，保存线程标示

• 释放锁时先判断线程标示是否与自己一致，一致则删除锁

  • 特性：

    • 利用set nx满足互斥性

    • 利用set ex保证发生异常时锁依然能释放，避免死锁，提高安全性

    • 利用Redis集群保证高可用和高并发特性

利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题我们开始是利用删之前 通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是我们没法保证拿锁比锁删锁是一个原子性的动作，最后通过Lua表达式来解决这个问题。?

10.??分布式锁 - Redission

基于SET NX实现的分布式锁存在下面的问题：

• 不可重入问题：可重入是指获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如Java中的synchronized和Lock锁都是可重入锁；而不可重入是指同一个线程无法多次获取同一把锁。
• 不可重试问题：是指目前的分布式锁只能尝试获取一次，是非阻塞式的，获取锁只尝试一次就返回，没有重试机制，我们认为合理的情况是：当线程在获得锁失败后，它应该能再次尝试获得锁。
• 超时释放：我们在加锁时增加了过期时间，这样我们可以防止死锁（锁超时释放可以避免死锁），但是如果业务卡顿的时间过长，也会导致锁释放，这样就会存在安全隐患问题（我的业务逻辑还没有执行完，锁就被释放了，这样就会导致锁被其它线程获取，相当于这本应该是我持有锁在执行业务逻辑的过程，但却被其它线程抢走了），虽然我们使用Lua表达式防止了删锁时候的误删问题，但是锁过期之后的续期问题我们还没有解决? =>? 没有解决锁自动过期导致其它线程也能上锁成功的问题。
• 主从一致性问题（主从模式即读写分离模式）：通过Redis加锁都是在某一个Redis集群节点上进行的，如果Redis提供了主从集群，主从同步存在延迟，当我们在向Master写数据时，主机需要异步的将数据同步给从机，而万一在数据同步过去之前，主机宕机了，就会出现数据不一致以及锁信息丢失的问题? ?=>? ?先在主节点获取到锁了，但是此时主节点数据还没完全同步到从节点，此时主节点挂了，然后发生故障转移从节点变成主节点了，这时候就会导致数据不一致，并且此时新的Master中实际上并没有锁信息，此时锁信息就已经丢掉了???=>? ?通过Redisson框架提供的RedLock算法实现分布式锁来解决。?

Redisson框架

• Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory? Data? Grid），它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其实就包含了各种分布式锁的实现。
• Redisson是一个基于Redis的Java客户端，它提供了丰富的功能，包括分布式锁的支持。?
• 目前最流行的Redis分布式锁就是Redisson了。

https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers

分布式锁-Redission快速入门?

• 基于Redisson可以非常简单的就获取一个可重入的分布式锁。?

1. 引入依赖：

<!--Redisson依赖-->
<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson</artifactId>
	<version>3.13.6</version>
</dependency>

2. 定义 / 配置Redisson客户端

package com.gch.redis.lock;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 定义或配置Redisson客户端
 */
@Configuration
public class RedissonConfig {
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redissonClient() {
        // 配置类
        Config config = new Config();
        // 添加Redis地址,这里添加了单点的地址,也可以使用config.useClusterServers()添加集群地址
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
        // 创建客户端 - 创建RedissonClient对象
        // config为Redisson的配置,Address为Redis地址,Password为Redis密码
        return Redisson.create(config);
    }
}

3. 接下来，在想要使用分布式锁的地方做如下调用即可：使用Redisson的分布式锁

    @Resource
    private RedissonClient redissonClient;

    @Test
    void testRedisson() throws InterruptedException {
        // 获取锁(可重入),指定锁的名称
        boolean result = redissonClient.getLock("anyLock")
                // 尝试获取锁,参数分别是:获取锁的最大等待时间(期间会重试),锁自动释放时间,时间单位
                .tryLock(10, 10, TimeUnit.SECONDS);
        // 判断获取锁是否成功
        if (!result) {
            log.error("获取锁失败...");
            return;
        }
        try {
            log.info("获取锁成功,执行业务逻辑...");
        } finally {
            // 释放锁
            redissonClient.getLock("anyLock").unlock();
            log.info("释放锁成功...");
        }
    }

    @Test
    void testRedisson2() throws InterruptedException {
        // 获取锁(可重入),指定锁的名称
        RLock lock = redissonClient.getLock("anyLock");
        // 尝试获取锁,参数分别是:获取锁的最大等待时间(期间会重试),锁自动释放时间,时间单位
        boolean result = lock.tryLock(10, 10, TimeUnit.SECONDS);
        // 判断获取锁是否成功
        if (!result) {
            log.error("获取锁失败...");
            return;
        }
        try {
            log.info("获取锁成功,执行业务逻辑...");
        } finally {
            // 释放锁
            lock.unlock();
            log.info("释放锁成功...");
        }
    }

11. Redisson的可重入锁原理

• 在Lock锁中，它是借助于底层的一个volatile的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state = 0，假如有人持有这把锁，那么 state = 1，如果持有这把锁的人再次持有这把锁，那么state就会 + 1；
• 如果是对于synchronized而言，它底层会有一个count，原理和state类似，也是重入一次就 + 1，释放一次就 - 1，直到减少成0时，表示当前这把锁没有被任何线程持有。?

我们自定义的分布式锁不能实现重入的原因是：

• 在获取锁的流程中，使用的是Redis的String数据类型，通过set命令来尝试获取锁，由于nx参数的存在，只有第一个尝试获取锁的线程能够成功，后续线程在尝试获取锁时会失败，因此，当一个线程连续两次尝试获取锁时（即锁的重入），第二次尝试获取锁的操作会失败，无法实现可重入。?

实现可重入锁的基本原理：

• 在获取锁的时候，当判断这个锁已经被线程持有的情况下，去看一下持有锁的线程是不是自己，如果是自己的话，会让自己再次获取锁，同时，还要有一个计数器去记录获取锁的重入次数，即总共获取了几次，每tryAcquire()成功获取一次计数值 + 1，将来在release释放锁的时候，计数值 - 1，这就是可重入锁的基本原理。?

如何实现可重入锁？

• 要实现可重入锁，需要在锁对象中记录获取锁的线程标识和重入次数，当一个线程尝试获取锁时，需要判断锁是否已经被该线程获取，如果是同一个线程获取锁，则重入次数 + 1，在释放锁时，需要判断重入次数是否为0，如果是0则删除锁，否则只是将重入次数 - 1（重入次数不为0说明还有其它业务没有处理完成），在实现可重入锁时，需要使用Hash代替之前的String字符串类型结果，因为需要在一个key里存储两个东西，key用来记录锁的名称，field用来记录线程标识，value位置用来记录锁的重入次数。
• 使用Hash结构之后我们必须手动使用exist命令判断锁是否存在，因为在哈希结构中没有nx这样的组合命令来判断锁是否存在，因此我们要将之前逻辑拆开，先判断锁是否存在，再手动通过expire命令设置过期时间。

使用Lua脚本来确保获取锁和释放锁的原子性。?

12. Redisson的锁可重试和WatchDog机制实现超时续约的原理

• 可重试：利用信号量和PubSub发布订阅功能实现等待、唤醒，获取锁失败的重试机制
• 利用WatchDog实现超时续约：客户端加锁的锁key默认生存时间leaseTime为30s，如果超过30s客户端还想一直持有这把锁，此时就要靠WatchDog自动延期机制了，只要客户端加锁成功，就会启动一个WatchDog看门狗，它是一个后台线程，会每隔10s检查一下，如果客户端还持有key，那么它就会重置超时时间，把锁的过期时间继续延长至30000毫秒 - 即30s，只要你这台服务实例没有挂掉，并且没有主动去释放锁，看门狗会每隔10s回你续约一下，去不断的延长key的生存时间，保证锁一直在你手中?=>? 利用Redisson框架提供的看门狗特性，可以在锁失效前不断延长过期时间，WatchDog机制解决锁续期问题（解决锁超时释放问题）

subscribe：订阅?

总结 - Redisson分布式锁底层的实现原理？???

• 可重入：利用hash结构记录线程ID和重入次数
• 可重试：利用信号量和Pub/Sub发布订阅功能实现等待、唤醒，获取锁失败的重试机制
• 超时续约：利用watchDog，每隔一段时间（releaseTime / 3），重置超时时间?

13. Redisson分布式锁解决主从一致性导致的锁失效问题

• Redisson采用MutiLock锁来解决主从一致性导致的锁失效问题，使用这把锁使得多个Redis节点变得独立了，节点之间没有主从关系，每个节点的地位都是一样的，从而避免了因为主从一致性问题导致的锁失效问题；
• 当其中有节点宕机时，其它节点仍然含有锁信息，其它节点仍然有效；
• 同时在获取锁的时候，需要在所有节点都获取重入锁成功才算获取锁成功，这样可以确保加锁的可靠性，虽然这种方法的运维成本较高，但是它是所有方案中最安全的一种。

使用MutiLock可以搭建主从同步集群 - 可以为每个节点去单独建立主从关系，进行主从同步，提高了灵活性和可用性：