黑马点评4

分布式锁

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。
分布式锁的核心思想就是让大家都使用同一把锁，只要大家使用的是同一把锁，那么我们就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路

常见的分布式锁有三种
Mysql：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少
Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用redis或者zookeeper作为分布式锁，利用setnx这个方法，如果插入key成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁
Zookeeper：zookeeper也是企业级开发中较好的一个实现分布式锁的方案。

实现思路

获取锁：
- 互斥：确保只能有一个线程获取锁
- 非阻塞：尝试一次，成功返回true，失败返回false
释放锁：
- 手动释放
- 超时释放：获取锁时添加一个超时时间

setnx lock thread1
expire lock 5

有可能在执行完第一句话的时候，服务器挂了，过期时间就无法执行，造成死锁的情况，无法保证原子性，因此我们想要这两个操作同时执行

可以使用下面这种方式，过期时间ex 为10s:

set lock thread ex 10 nx

实现分布式锁

锁的基本接口：

public interface ILock {

    /**
     * 尝试获取锁
     * @param timeoutSec 超时时间，单位秒 过期自动释放锁
     * @return true 获取成功，false 获取失败
     */
    boolean tryLock(long timeoutSec);
    
    /**
     * 释放锁
     */
    void unlock();
}

实现类：

public class SimpleRedisLock implements ILock {

    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    public static final String KEY_PREFIX = "lock:";

    @Override
    public boolean tryLock(long timeoutSec) {
        //获取当前线程的id
        long threadId = Thread.currentThread().getId();
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, String.valueOf(threadId),
                        timeoutSec, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(success);
    }

    @Override
    public void unlock() {
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
}

修改业务代码：
原来有问题的代码：

        Long userId = UserHolder.getUser().getId();
        synchronized (userId.toString().intern()) {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }

修改如下：

        Long userId = UserHolder.getUser().getId();

        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        boolean isLock = lock.tryLock(5);
        if (!isLock) {
            //获取失败，返回错误或者 重试
            return Result.fail("服务器繁忙");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            lock.unlock();
        }

Redis分布式锁误删

误删问题：
持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2来尝试获得锁，就拿到了这把锁，然后线程2在持有锁执行过程中，线程1反应过来，继续执行，而线程1执行过程中，走到了删除锁逻辑，此时就会把本应该属于线程2的锁进行删除，这就是误删别人锁的情况说明
解决方案：
解决方案就是在每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁。

解决：
在获取锁时存入线程标示（可以用UUID表示）在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致

代码：

	public static final String KEY_PREFIX = "lock:";
    public static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";

    @Override
    public boolean tryLock(long timeoutSec) {
        //获取当前线程的id
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, String.valueOf(threadId),
                        timeoutSec, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(success);
    }

    @Override
    public void unlock() {
        //获取当前线程的id
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        //获取锁的值
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        if (threadId.equals(id)) {
            //一致，说明是当前线程的锁，删除
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }

分布式锁原子性问题

线程1现在持有锁之后，在执行业务逻辑过程中，他正准备删除锁，而且已经走到了条件判断的过程中，比如他已经拿到了当前这把锁确实是属于他自己的，正准备删除锁，但是此时他的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，我们要防止刚才的情况发生，

Lua脚本解决多条命令原子性问题

lua脚本可以编写多条redis命令，确保多条命令执行时的原子性.
调用函数

redis.call("命令名称 ","key","其他参数 ")

例如：

-- 执行set name jack  
redis.call("set","name","jack")

先执行set name Rose，再执行get name，则脚本如下：

--先执行set name Rose，再执行get name，则脚本如下：
redis.call("set","name","Rose")
local name = redis.call("get", "name")
return name

释放锁的业务流程:
1、获取锁中的线程标示
2、判断是否与指定的标示（当前线程标示）一致
3、如果一致则释放锁（删除）
4、如果不一致则什么都不做

Lua脚本如下：

if (redis.call('GET', KEYS[1]) == ARGV[1]) then  
-- 一致，则删除锁  
return redis.call('DEL', KEYS[1])  
end  
-- 不一致，则直接返回  
return 0

代码：

    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }

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

分布式锁Redission

Redisson是一个在Redis的基础上实现的Java驻内存数据网格。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

添加依赖：

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

配置客户端：

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient()   {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://localhost:6379")
                .setPassword("123456");
        return Redisson.create(config);
    }
}

一般的使用步骤：

@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();
        }
    }
}

修改原来的代码：

//        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        boolean isLock = lock.tryLock();

分布式锁-redission可重入锁原理

在Lock锁中，他是借助于底层的一个voaltile的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state=0，假如有人持有这把锁，那么state=1，如果持有这把锁的人再次持有这把锁，那么state就会+1 ，在Java的内建锁（synchronized）中，重入锁的状态通常由一个count变量来表示，每次锁被获得时，count会增加1，每次锁被释放时，count会减少1。当count减少到0时，表示没有线程持有锁。这样，同一个线程可以多次获得锁，而不会导致死锁。

在Redisson中，可重入锁是通过在Redis上使用特定的数据结构和Lua脚本来实现的。

local lockExists = redis.call("exists", KEYS[1]) -- 检查锁是否存在

if lockExists == 0 then
    redis.call("hset", KEYS[1], ARGV[1], 1) -- 锁不存在，创建锁，并将拥有者设置为1
    redis.call("pexpire", KEYS[1], ARGV[2]) -- 设置锁的过期时间
    return 1 -- 返回1表示成功获取锁
end

local lockOwner = redis.call("hget", KEYS[1], ARGV[1]) -- 获取锁的拥有者

if lockOwner == false then
    redis.call("hset", KEYS[1], ARGV[1], 1) -- 拥有者不存在，创建锁，并将拥有者设置为1
    redis.call("pexpire", KEYS[1], ARGV[2]) -- 设置锁的过期时间
    return 1 -- 返回1表示成功获取锁
end

if lockOwner == ARGV[1] then
    local counter = redis.call("hincrby", KEYS[1], ARGV[1], 1) -- 拥有者是当前线程，增加拥有次数
    redis.call("pexpire", KEYS[1], ARGV[2]) -- 更新锁的过期时间
    return counter -- 返回拥有次数
end

return 0 -- 返回0表示获取锁失败

这个Lua脚本的作用是：

首先检查大键表示的锁是否存在。如果锁不存在，则创建锁，并将小键表示的拥有者设置为1，并为锁设置过期时间。返回1表示成功获取锁。
如果锁存在，再检查小键表示的锁的拥有者是不是当前线程（线程标识通过ARGV[1]传递）。如果是当前线程，增加拥有次数，并更新锁的过期时间。返回拥有次数。
如果锁已经被其他线程拥有，返回0表示获取锁失败。

Redission锁重试机制
Redisson提供了重试机制来处理获取分布式锁时的竞争条件。这个机制可以用于在获取锁失败时，尝试多次获取锁，以减少竞争。
Redission的锁重试机制是通过org.redisson.RedissonRedLock类中的tryLockInner方法实现的。这个方法是RedLock分布式锁的内部实现，它通过Lua脚本与Redis服务器交互以获取锁。

查找tryLockInner方法：在Redission的源码中，可以找到RedissonRedLock类，然后查找tryLockInner方法。这个方法是用来尝试获取锁的核心逻辑。
Lua脚本：在tryLockInner方法中，使用Lua脚本来与Redis服务器进行交互。这个Lua脚本实现了锁的获取逻辑，包括了锁的存在检查、锁的创建、锁的续约等等。
重试机制：在获取锁的过程中，看到有关重试的逻辑。这包括了重试次数和重试间隔的控制，通常使用retryAttempts和retryInterval参数来配置。
锁的状态：在锁的重试机制中，可能需要跟踪锁的状态，以确定是否已经成功获取锁。你会看到一些变量或标志来表示锁的状态。
超时处理：还会有关于等待锁的超时时间处理，如果等待超过一定时间仍然无法获取锁，会抛出异常。
与Redis交互：Redission通过Redisson客户端与Redis服务器进行交互，一些Redis命令的调用，如set、eval等，用来实现锁的获取和续约。

看门狗机制：
在分布式系统和分布式锁中，"看门狗"（也称为"锁的续租"或"锁的续约"机制）是一种用于确保锁的有效性和持续性的机制。这个机制的目的是防止因为锁持有者在持有锁期间发生故障或长时间处理任务而导致锁无法释放的情况。
看门狗机制的基本工作方式如下：

当一个线程成功获取锁时，它会同时启动一个定时器或计时器，设置一个锁的持续时间。这个持续时间通常是一个较短的时间段，比如锁的过期时间的一半。
在锁的持续时间内，锁的拥有者需要周期性地“喂狗”，也就是不断重置或续租锁的持续时间。这通常是通过向锁存储中更新锁的时间戳或其他信息来实现的。
如果锁的拥有者在持续时间内没有续租锁，比如因为线程崩溃或异常退出，那么锁将自动过期，其他线程将有机会尝试获取锁。
看门狗机制的好处在于它能够防止锁被永久地占用。即使锁的拥有者在某些情况下无法释放锁，也会在锁的持续时间过期后，使其他线程有机会获取锁，避免了死锁或长时间阻塞的情况。
Redisson等一些分布式锁实现库使用了看门狗机制来支持锁的续租。通过定期续租锁，可以确保锁不会因为拥有者的故障而永久丧失。如果锁的拥有者能够定期执行续租操作，锁可以一直保持有效。如果锁的拥有者无法续租，锁将在过期时间后自动释放，从而确保其他线程有机会获取锁。
原理图如下：

redission锁的MutiLock原理

为了提高redis的可用性，我们会搭建集群或者主从，现在以主从为例
此时我们去写命令，写在主机上，主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的，这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。

黑马点评4

# 黑马点评4

# 分布式锁

# 实现思路

# 实现分布式锁

# Redis分布式锁误删

# 分布式锁原子性问题

# Lua脚本解决多条命令原子性问题

# 分布式锁Redission

# 分布式锁-redission可重入锁原理

# redission锁的MutiLock原理

黑马点评4

分布式锁

实现思路

实现分布式锁

Redis分布式锁误删

分布式锁原子性问题

Lua脚本解决多条命令原子性问题

分布式锁Redission

分布式锁-redission可重入锁原理

redission锁的MutiLock原理