【微服务组件之分布式锁】分布式锁

如果在一个分布式系统中，我们从数据库中读取一个数据，然后修改保存，这种情况很容易遇到并发问题。因为读取和更新保存不是一个原子操作，在并发时就会导致数据的不正确。这种场景其实并不少见，比如电商秒杀活动，库存数量的更新就会遇到。如果是单机应用，直接使用本地锁就可以避免。如果是分布式应用，本地锁派不上用场，这时就需要引入

分布式锁

来解决。

由此可见分布式锁的目的其实很简单，就是

为了保证多台服务器在执行某一段代码时保证只有一台服务器执行

。

在这里插入图片描述

为了保证分布式锁的可用性，至少要确保锁的实现要同时满足以下几点：

互斥性。在任何时刻，保证只有一个客户端持有锁。 -锁的原子性
不能出现死锁。如果在一个客户端持有锁的期间，这个客户端崩溃了，也要保证后续的其他客户端可以上锁。 –设置锁超时
保证上锁和解锁都是同一个

客户端

。 -识别锁持有者

一般来说，实现分布式锁的方式有以下几种：

使用MySQL，基于唯一索引。基本方案
使用ZooKeeper，基于临时有序节点。
使用Redis，基于setnx命令

。-主流方案

数据库锁

数据库悲观锁

所谓悲观锁，悲观锁是对数据被的修改持悲观态度（认为数据在被修改的时候一定会存在并发问题），因此在整个数据处理过程中将数据锁定。

悲观锁的实现，往往依靠数据库提供的锁机制（也只有数据库层提供的锁机制才能真正保证数据访问的排他性，否则，即使在应用层中实现了加锁机制，也无法保证外部系统不会修改数据）。

数据库的行锁、表锁、排他锁等都是悲观锁，这里以行锁为例，进行介绍。以我们常用的MySQL为例，我们通过使用select…for update语句, 执行该语句后，会在表上加持行锁，一直到事务提交，解除行锁。

使用场景举例:

在秒杀案例中，生成订单和扣减库存的操作，可以通过商品记录的行锁，进行保护。们通过使用select…for update语句，在查询商品表库存时将该条记录加锁，待下单减库存完成后，再释放锁。

示例的SQL如下：

//0.开始事务
begin; 
	
//1.查询出商品信息

select stockCount from seckill_good where id=1 for update;

//2.根据商品信息生成订单

insert into seckill_order (id,good_id) values (null,1);

//3.修改商品stockCount减一

update seckill_good set stockCount=stockCount-1 where id=1;

//4.提交事务

commit;

以上，在对id = 1的记录修改前，先通过for update的方式进行加锁，然后再进行修改。这就是比较典型的悲观锁策略。

如果以上修改库存的代码发生并发，同一时间只有一个线程可以开启事务并获得id=1的锁，其它的事务必须等本次事务提交之后才能执行。这样我们可以保证当前的数据不会被其它事务修改。

我们使用select_for_update，另外一定要写在事务中.

注意：要使用悲观锁，我们必须关闭mysql数据库中自动提交的属性，命令set autocommit=0;即可关闭，因为MySQL默认使用autocommit模式，也就是说，当你执行一个更新操作后，MySQL会立刻将结果进行提交。

悲观锁的实现，往往依靠数据库提供的锁机制。在数据库中，悲观锁的流程如下：

在对记录进行修改前，先尝试为该记录加上排他锁（exclusive locking）。
如果加锁失败，说明该记录正在被修改，那么当前查询可能要等待或者抛出异常。具体响应方式由开发者根据实际需要决定。
如果成功加锁，那么就可以对记录做修改，事务完成后就会解锁了。
其间如果有其他事务对该记录做加锁的操作，都要等待当前事务解锁或直接抛出异常。

数据库乐观锁

使用乐观锁就不需要借助数据库的锁机制了。

乐观锁的概念中其实已经阐述了他的具体实现细节：主要就是两个步骤：冲突检测和数据更新。其实现方式有一种比较典型的就是Compare and Swap(CAS)技术。

CAS是项乐观锁技术，当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。

CAS的实现中，在表中增加一个version字段，操作前先查询version信息，在数据提交时检查version字段是否被修改，如果没有被修改则进行提交，否则认为是过期数据。

比如前面的扣减库存问题，通过乐观锁可以实现如下：

//1.查询出商品信息
            
select stockCount, version from seckill_good where id=1;
            
//2.根据商品信息生成订单
insert into seckill_order (id,good_id) values (null,1);

//3.修改商品库存
update seckill_good set stockCount=stockCount-1, version = version+1 where id=1, version=version;

以上，我们在更新之前，先查询一下库存表中当前版本（version），然后在做update的时候，以version 作为一个修改条件。

当我们提交更新的时候，判断数据库表对应记录的当前version与第一次取出来的version进行比对，如果数据库表当前version与第一次取出来的version相等，则予以更新，否则认为是过期数据。

CAS 乐观锁有两个问题：

(1) CAS 存在一个比较重要的问题，即ABA问题. 解决的办法是version字段顺序递增。

(2) 乐观锁的方式，在高并发时，只有一个线程能执行成功，会造成大量的失败，这给用户的体验显然是很不好的。

Redis实现思路

本文重点介绍Redis分布式锁，分为两个维度进行介绍：

（1）基于Jedis手工造轮子分布式锁

（2）介绍Redission 分布式锁的使用和原理。

分布式锁一般有如下的特点：

互斥性：同一时刻只能有一个线程持有锁

可重入性：同一节点上的同一个线程如果获取了锁之后能够再次获取锁

锁超时：和J.U.C中的锁一样支持锁超时，防止死锁

高性能和高可用：加锁和解锁需要高效，同时也需要保证高可用，防止分布式锁失效

具备阻塞和非阻塞性：能够及时从阻塞状态中被唤醒

手工造轮子：基于Jedis 的API实现分布式锁

我们首先讲解 Jedis 普通分布式锁实现，并且是纯手工的模式，从最为基础的Redis命令开始。

只有充分了解与分布式锁相关的普通Redis命令，才能更好的了解高级的Redis分布式锁的实现，因为高级的分布式锁的实现完全基于普通Redis命令。

Redis几种架构

Redis发展到现在，几种常见的部署架构有

单机模式；
主从模式；
哨兵模式；
集群模式；

从分布式锁的角度来说，无论是单机模式、主从模式、哨兵模式、集群模式，其原理都是类同的。只是主从模式、哨兵模式、集群模式的更加的高可用、或者更加高并发。

Redis分布式锁机制，主要借助setnx和expire两个命令完成。

setnx命令:

SETNX 是SET if Not eXists的简写。将 key 的值设为 value，当且仅当 key 不存在; 若给定的 key 已经存在，则 SETNX 不做任何动作。

下面为客户端使用示例：

127.0.0.1:6379> set lock "unlock"
OK
127.0.0.1:6379> setnx lock "unlock"
(integer) 0
127.0.0.1:6379> setnx lock "lock"
(integer) 0
127.0.0.1:6379>

expire命令:

expire命令为 key 设置生存时间，当 key 过期时(生存时间为 0 )，它会被自动删除. 其格式为：

EXPIRE key seconds

下面为客户端使用示例：

127.0.0.1:6379> expire lock 10
(integer) 1
127.0.0.1:6379> ttl lock
8

基于Jedis API的分布式锁的总体流程：

通过Redis的setnx、expire命令可以实现简单的锁机制：

key不存在时创建，并设置value和过期时间，返回值为1；成功获取到锁；
如key存在时直接返回0，抢锁失败；
持有锁的线程释放锁时，手动删除key；或者过期时间到，key自动删除，锁释放。
线程调用setnx方法成功返回1认为加锁成功，其他线程要等到当前线程业务操作完成释放锁后，才能再次调用setnx加锁成功。

在这里插入图片描述

以上简单redis分布式锁的问题：

如果出现了这么一个问题：如果
setnx
是成功的，但是
expire
设置失败，一旦出现了释放锁失败，或者没有手工释放，那么这个锁永远被占用，其他线程永远也抢不到锁。

所以,需要保障setnx和expire两个操作的原子性，要么全部执行，要么全部不执行，二者不能分开。

解决的办法有两种：

使用set的命令时，同时设置过期时间，不再单独使用 expire命令
使用lua脚本，将加锁的命令放在lua脚本中原子性的执行

简单加锁：使用set的命令时，同时设置过期时间

使用set的命令时，同时设置过期时间的示例如下：

127.0.0.1:6379> set unlock "234" EX 100 NX
(nil)
127.0.0.1:6379> 
127.0.0.1:6379> set test "111" EX 100 NX
OK

这样就完美的解决了分布式锁的原子性； set 命令的完整格式：

set key value [EX seconds] [PX milliseconds] [NX|XX]


EX seconds：设置失效时长，单位秒
PX milliseconds：设置失效时长，单位毫秒
NX：key不存在时设置value，成功返回OK，失败返回(nil)
XX：key存在时设置value，成功返回OK，失败返回(nil)

使用set命令实现加锁操作，先展示加锁的简单代码实习，再带大家慢慢解释为什么这样实现。

加锁的简单代码实现

package com.crazymaker.springcloud.standard.lock;


@Slf4j
@Data
@AllArgsConstructor
public class JedisCommandLock {

    private  RedisTemplate redisTemplate;

    private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";

    /**
     * 尝试获取分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁
     * @param requestId 请求标识
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public static   boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {

        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);

        if (LOCK_SUCCESS.equals(result)) {
            return true;
        }
        return false;

    }

  

}

可以看到，我们加锁用到了

Jedis的set Api

：

这个set()方法一共有五个形参：

– 第一个为key，我们使用key来当锁，因为key是唯一的。

– 第二个为value，我们传的是requestId，很多童鞋可能不明白，有key作为锁不就够了吗，为什么还要用到value？原因就是我们在上面讲到可靠性时，分布式锁要满足第四个条件解铃还须系铃人，通过给value赋值为requestId，我们就知道这把锁是哪个请求加的了，在解锁的时候就可以有依据。

requestId可以使用UUID.randomUUID().toString()方法生成。

– 第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作；

– 第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定。

– 第五个为time，与第四个参数相呼应，代表key的过期时间。

总的来说，执行上面的set()方法就只会导致两种结果：

当前没有锁（key不存在），那么就进行加锁操作，并对锁设置个有效期，同时value表示加锁的客户端。
已有锁存在，不做任何操作。-

心细的童鞋就会发现了，我们的加锁代码满足前面描述的四个条件中的三个。

首先，set()加入了NX参数，可以保证如果已有key存在，则函数不会调用成功，也就是只有一个客户端能持有锁，满足互斥性。
其次，由于我们对锁设置了过期时间，即使锁的持有者后续发生崩溃而没有解锁，锁也会因为到了过期时间而自动解锁（即key被删除），不会被永远占用（而发生死锁）。
最后，因为我们将value赋值为requestId，代表加锁的客户端请求标识，那么在客户端在解锁的时候就可以进行校验是否是同一个客户端。

由于我们只考虑Redis单机部署的场景，所以容错性我们暂不考虑。

基于Jedis 的API实现简单解锁代码

还是先展示代码，再带大家慢慢解释为什么这样实现。

解锁的简单代码实现

：

package com.crazymaker.springcloud.standard.lock;


@Slf4j
@Data
@AllArgsConstructor
public class JedisCommandLock {

  

    private static final Long RELEASE_SUCCESS = 1L;

    /**
     * 释放分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁
     * @param requestId 请求标识
     * @return 是否释放成功
     */
    public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {

        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));

        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;

    }

}

那么这段Lua代码的功能是什么呢？

其实很简单，首先获取锁对应的value值，检查是否与requestId相等，如果相等则删除锁（解锁）。

第一行代码，我们写了一个简单的Lua脚本代码。

第二行代码，我们将Lua代码传到jedis.eval()方法里，并使参数KEYS[1]赋值为lockKey，ARGV[1]赋值为requestId。eval()方法是将Lua代码交给Redis服务端执行。

那么为什么要使用Lua语言来实现呢？

因为要确保上述操作是原子性的。那么为什么执行eval()方法可以确保原子性，源于Redis的特性.

简单来说，就是在eval命令执行Lua代码的时候，Lua代码将被当成一个命令去执行，并且直到eval命令执行完成，Redis才会执行其他命令

错误示例1

最常见的解锁代码就是直接使用 jedis.del() 方法删除锁，这种不先判断锁的拥有者而直接解锁的方式，会导致任何客户端都可以随时进行解锁，即使这把锁不是它的。

public static void wrongReleaseLock1(Jedis jedis, String lockKey) {

jedis.del(lockKey);

}

错误示例2

这种解锁代码乍一看也是没问题，甚至我之前也差点这样实现，与正确姿势差不多，唯一区别的是分成两条命令去执行，代码如下：

public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {
        
    // 判断加锁与解锁是不是同一个客户端
    if (requestId.equals(jedis.get(lockKey))) {
        // 若在此时，这把锁突然不是这个客户端的，则会误解锁
        jedis.del(lockKey);
    }

再造轮子：基于Lua脚本实现分布式锁

lua脚本的好处

前面提到，在redis中执行lua脚本，有如下的好处：

那么为什么要使用Lua语言来实现呢？

因为要确保上述操作是原子性的。那么为什么执行eval()方法可以确保原子性，源于Redis的特性.

简单来说，就是在eval命令执行Lua代码的时候，Lua代码将被当成一个命令去执行，并且直到eval命令执行完成，Redis才会执行其他命

所以：

大部分的开源框架（如 redission）中的分布式锁组件，都是用纯lua脚本实现的。

题外话： lua脚本是

高并发、高性能的必备脚本语言

基于纯Lua脚本的分布式锁的执行流程

加锁和删除锁的操作，使用纯lua进行封装，保障其执行时候的原子性。

基于纯Lua脚本实现分布式锁的执行流程，大致如下：

在这里插入图片描述

加锁的Lua脚本： lock.lua

--- -1 failed
--- 1 success
---
local key = KEYS[1]
local requestId = KEYS[2]
local ttl = tonumber(KEYS[3])
local result = redis.call('setnx', key, requestId)
if result == 1 then
    --PEXPIRE:以毫秒的形式指定过期时间
    redis.call('pexpire', key, ttl)
else
    result = -1;
    -- 如果value相同，则认为是同一个线程的请求，则认为重入锁
    local value = redis.call('get', key)
    if (value == requestId) then
        result = 1;
        redis.call('pexpire', key, ttl)
    end
end
--  如果获取锁成功，则返回 1
return result

解锁的Lua脚本： unlock.lua：

--- -1 failed
--- 1 success

-- unlock key
local key = KEYS[1]
local requestId = KEYS[2]
local value = redis.call('get', key)
if value == requestId then
    redis.call('del', key);
    return 1;
end
return -1

两个文件，放在资源文件夹下备用：

在这里插入图片描述

在Java中调用lua脚本，完成加锁操作

下一步，实现Lock接口, 完成JedisLock的分布式锁。

其加锁操作，通过调用 lock.lua脚本完成，代码如下：

package com.crazymaker.springcloud.standard.lock;

import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

@Slf4j
@Data
@AllArgsConstructor
public class JedisLock implements Lock {

    private RedisTemplate redisTemplate;

    RedisScript<Long> lockScript = null;
    RedisScript<Long> unLockScript = null;

    public static final int DEFAULT_TIMEOUT = 2000;
    public static final Long LOCKED = Long.valueOf(1);
    public static final Long UNLOCKED = Long.valueOf(1);
    public static final Long WAIT_GAT = Long.valueOf(200);
    public static final int EXPIRE = 2000;


    String key;
    String lockValue;  // lockValue 锁的value ,代表线程的uuid

    /**
     * 默认为2000ms
     */
    long expire = 2000L;

    public JedisLock(String lockKey, String lockValue) {
        this.key = lockKey;
        this.lockValue = lockValue;
    }

    private volatile boolean isLocked = false;

    private Thread thread;

    /**
     * 获取一个分布式锁 , 超时则返回失败
     *
     * @return 获锁成功 - true | 获锁失败 - false
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {

        //本地可重入
        if (isLocked && thread == Thread.currentThread()) {
            return true;
        }
        expire = unit != null ? unit.toMillis(time) : DEFAULT_TIMEOUT;
        long startMillis = System.currentTimeMillis();
        Long millisToWait = expire;

        boolean localLocked = false;

        int turn = 1;
        while (!localLocked) {

            localLocked = this.lockInner(expire);
            if (!localLocked) {
                millisToWait = millisToWait - (System.currentTimeMillis() - startMillis);
                startMillis = System.currentTimeMillis();
                if (millisToWait > 0L) {
                    /**
                     * 还没有超时
                     */
                    ThreadUtil.sleepMilliSeconds(WAIT_GAT);
                    log.info("睡眠一下，重新开始，turn:{},剩余时间：{}", turn++, millisToWait);
                } else {
                    log.info("抢锁超时");
                    return false;
                }
            } else {
                isLocked = true;
                localLocked = true;
            }
        }
        return isLocked;
    }


  
    /**
     * 有返回值的抢夺锁
     *
     * @param millisToWait
     */
    public boolean lockInner(Long millisToWait) {
        if (null == key) {
            return false;
        }
        try {
            List<String> redisKeys = new ArrayList<>();
            redisKeys.add(key);
            redisKeys.add(lockValue);
            redisKeys.add(String.valueOf(millisToWait));
            Long res = (Long) redisTemplate.execute(lockScript, redisKeys);

            return res != null && res.equals(LOCKED);
        } catch (Exception e) {
            e.printStackTrace();
            throw BusinessException.builder().errMsg("抢锁失败").build();
        }

    }

   
}

在Java中调用lua脚本，完成解锁操作

其解锁操作，通过调用unlock.lua脚本完成，代码如下：

package com.crazymaker.springcloud.standard.lock;

import com.crazymaker.springcloud.common.exception.BusinessException;
import com.crazymaker.springcloud.common.util.ThreadUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

@Slf4j
@Data
@AllArgsConstructor
public class JedisLock implements Lock {

    private RedisTemplate redisTemplate;

    RedisScript<Long> lockScript = null;
    RedisScript<Long> unLockScript = null;

    //释放锁
    @Override
    public void unlock() {
        if (key == null || requestId == null) {
            return;
        }
        try {
            List<String> redisKeys = new ArrayList<>();
            redisKeys.add(key);
            redisKeys.add(requestId);
            Long res = (Long) redisTemplate.execute(unLockScript, redisKeys);

        } catch (Exception e) {
            e.printStackTrace();
            throw BusinessException.builder().errMsg("释放锁失败").build();

        }
    }
  
   
}

编写RedisLockService用于管理JedisLock

编写个分布式锁服务，用于加载lua脚本，创建分布式锁，代码如下：

package com.crazymaker.springcloud.standard.lock;

import com.crazymaker.springcloud.common.util.IOUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

@Slf4j
@Data
public class RedisLockService
{

    private RedisTemplate redisTemplate;

    static String lockLua = "script/lock.lua";
    static String unLockLua = "script/unlock.lua";
    static RedisScript<Long> lockScript = null;
    static RedisScript<Long> unLockScript = null;
    {
        String script = IOUtil.loadJarFile(RedisLockService.class.getClassLoader(),lockLua);
//        String script = FileUtil.readString(lockLua, Charset.forName("UTF-8" ));
        if(StringUtils.isEmpty(script))
        {
            log.error("lua load failed:"+lockLua);
        }

        lockScript = new DefaultRedisScript<>(script, Long.class);

//        script = FileUtil.readString(unLockLua, Charset.forName("UTF-8" ));
        script =  IOUtil.loadJarFile(RedisLockService.class.getClassLoader(),unLockLua);
        if(StringUtils.isEmpty(script))
        {
            log.error("lua load failed:"+unLockLua);
        }
        unLockScript = new DefaultRedisScript<>(script, Long.class);

    }

    public RedisLockService(RedisTemplate redisTemplate)
    {
        this.redisTemplate = redisTemplate;
    }


    public Lock getLock(String lockKey, String lockValue) {
        JedisLock lock=new JedisLock(lockKey,lockValue);
        lock.setRedisTemplate(redisTemplate);
        lock.setLockScript(lockScript);
        lock.setUnLockScript(unLockScript);
        return lock;
    }
}

测试用例

接下来，终于可以上测试用例了

package com.crazymaker.springcloud.lock;

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = {DemoCloudApplication.class})
// 指定启动类
public class RedisLockTest {

    @Resource
    RedisLockService redisLockService;

    private ExecutorService pool = Executors.newFixedThreadPool(10);

    @Test
    public void testLock() {
        int threads = 10;
        final int[] count = {0};
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                String lockValue = UUID.randomUUID().toString();

                try {
                    Lock lock = redisLockService.getLock("test:lock:1", lockValue);
                    boolean locked = lock.tryLock(10, TimeUnit.SECONDS);

                    if (locked) {
                        for (int j = 0; j < 1000; j++) {
                            count[0]++;
                        }

                        log.info("count = " + count[0]);
                        lock.unlock();
                    } else {
                        System.out.println("抢锁失败");
                    }


                } catch (Exception e) {
                    e.printStackTrace();
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("10个线程每个累加1000为： = " + count[0]);
        //输出统计结果
        float time = System.currentTimeMillis() - start;

        System.out.println("运行的时长为(ms)：" + time);
        System.out.println("每一次执行的时长为(ms)：" + time / count[0]);

    }

}

执行用例，结果如下：

2021-05-04 23:02:11.900  INFO 22120 --- [pool-1-thread-7] c.c.springcloud.lock.RedisLockTest       LN:50 count = 6000
2021-05-04 23:02:11.901  INFO 22120 --- [pool-1-thread-1] c.c.springcloud.standard.lock.JedisLock  LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:11.902  INFO 22120 --- [pool-1-thread-1] c.c.springcloud.lock.RedisLockTest       LN:50 count = 7000
2021-05-04 23:02:12.100  INFO 22120 --- [pool-1-thread-4] c.c.springcloud.standard.lock.JedisLock  LN:81 睡眠一下，重新开始，turn:3,剩余时间：9586
2021-05-04 23:02:12.101  INFO 22120 --- [pool-1-thread-5] c.c.springcloud.standard.lock.JedisLock  LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:12.101  INFO 22120 --- [pool-1-thread-8] c.c.springcloud.standard.lock.JedisLock  LN:81 睡眠一下，重新开始，turn:3,剩余时间：9585
2021-05-04 23:02:12.101  INFO 22120 --- [pool-1-thread-4] c.c.springcloud.lock.RedisLockTest       LN:50 count = 8000
2021-05-04 23:02:12.102  INFO 22120 --- [pool-1-thread-8] c.c.springcloud.lock.RedisLockTest       LN:50 count = 9000
2021-05-04 23:02:12.304  INFO 22120 --- [pool-1-thread-5] c.c.springcloud.standard.lock.JedisLock  LN:81 睡眠一下，重新开始，turn:4,剩余时间：9383
2021-05-04 23:02:12.307  INFO 22120 --- [pool-1-thread-5] c.c.springcloud.lock.RedisLockTest       LN:50 count = 10000
10个线程每个累加1000为： = 10000
运行的时长为(ms)：827.0
每一次执行的时长为(ms)：0.0827

STW导致的锁过期问题

下面有一个简单的使用锁的例子，在10秒内占着锁：

  //写数据到文件
function writeData(filename, data) {
    boolean locked = lock.tryLock(10, TimeUnit.SECONDS);
    if (!locked) {
        throw 'Failed to acquire lock';
    }

    try {
        //将数据写到文件
        var file = storage.readFile(filename);
        var updated = updateContents(file, data);
        storage.writeFile(filename, updated);
    } finally {
        lock.unlock();
    }
}

问题是：如果在写文件过程中，发生了 fullGC，并且其时间跨度较长，超过了10秒，那么，分布式就自动释放了。

在此过程中，client2 抢到锁，写了文件。

client1 的fullGC完成后，也继续写文件，

注意，此时client1 的并没有占用锁

，此时写入会导致文件数据错乱，发生线程安全问题。

这就是STW导致的锁过期问题。

STW导致的锁过期问题，具体如下图所示：

STW导致的锁过期问题,大概的解决方案，有：

1：模拟CAS乐观锁的方式，增加版本号

2：

watch dog自动延期机制

1：模拟CAS乐观锁的方式，增加版本号（如下图中的token）

此方案如果要实现，需要调整业务逻辑，与之配合，所以会入侵代码。

2：watch dog自动延期机制

客户端1加锁的锁key默认生存时间才30秒，如果超过了30秒，客户端1还想一直持有这把锁，怎么办呢？

简单！只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会每隔10秒检查一下，如果客户端1还持有锁key，那么就会不断的延长锁key的生存时间。

redission，采用的就是这种方案，此方案不会入侵业务代码。

为啥推荐使用Redission

作为 Java 开发人员，我们若想在程序中集成 Redis，必须使用 Redis 的第三方库。目前大家使用的最多的第三方库是jedis。

和SpringCloud gateway一样，Redisson也是基于Netty实现的，是更高性能的第三方库。所以，这里推荐大家使用Redission替代 jedis。

Redisson简介

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还实现了可重入锁（Reentrant Lock）、公平锁（Fair Lock、联锁（MultiLock）、红锁（RedLock）、读写锁（ReadWriteLock）等，还提供了许多分布式服务。

Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

Redisson与Jedis的对比

1.概况对比

Jedis是Redis的java实现的客户端，其API提供了比较全面的的Redis命令的支持，Redisson实现了分布式和可扩展的的java数据结构，和Jedis相比，功能较为简单，不支持字符串操作，不支持排序，事物，管道，分区等Redis特性。Redisson的宗旨是促进使用者对Redis的关注分离，从而让使用者能够将精力更集中的放在处理业务逻辑上。

2.可伸缩性

Jedis使用阻塞的I/O，且其方法调用都是同步的，程序流程要等到sockets处理完I/O才能执行，不支持异步，Jedis客户端实例不是线程安全的，所以需要通过连接池来使用Jedis。

Redisson使用非阻塞的I/O和基于Netty框架的事件驱动的通信层，其方法调用时异步的。Redisson的API是线程安全的，所以操作单个Redisson连接来完成各种操作。

3.第三方框架整合

Redisson在Redis的基础上实现了java缓存标准规范；Redisson还提供了Spring Session回话管理器的实现。

Redission 的源码地址：

官网： https://redisson.org/

github： https://github.com/redisson/redisson#quick-start

特性 & 功能：

支持 Redis 单节点（single）模式、哨兵（sentinel）模式、主从（Master/Slave）模式以及集群（Redis Cluster）模式

程序接口调用方式采用异步执行和异步流执行两种方式

数据序列化，Redisson 的对象编码类是用于将对象进行序列化和反序列化，以实现对该对象在 Redis 里的读取和存储

单个集合数据分片，在集群模式下，Redisson 为单个 Redis 集合类型提供了自动分片的功能

提供多种分布式对象，如：Object Bucket，Bitset，AtomicLong，Bloom Filter 和 HyperLogLog 等

提供丰富的分布式集合，如：Map，Multimap，Set，SortedSet，List，Deque，Queue 等

分布式锁和同步器的实现，可重入锁（Reentrant Lock），公平锁（Fair Lock），联锁（MultiLock），红锁（Red Lock），信号量（Semaphonre），可过期性信号锁（PermitExpirableSemaphore）等

提供先进的分布式服务，如分布式远程服务（Remote Service），分布式实时对象（Live Object）服务，分布式执行服务（Executor Service），分布式调度任务服务（Schedule Service）和分布式映射归纳服务（MapReduce）

Redisson的使用

如何安装 Redisson

安装 Redisson 最便捷的方法是使用 Maven 或者 Gradle：

•Maven

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

•Gradle

compile group: 'org.redisson', name: 'redisson', version: '3.11.4'

目前 Redisson 最新版是 3.11.4，当然你也可以通过搜索 Maven 中央仓库 mvnrepository[1] 来找到 Redisson 的各种版本。

获取RedissonClient对象

RedissonClient有多种模式，主要的模式有：

单节点模式
哨兵模式
主从模式
集群模式

首先介绍单节点模式。

单节点模式的程序化配置方法，大致如下：

Config config = new Config();
config.useSingleServer().setAddress("redis://myredisserver:6379");
RedissonClient redisson = Redisson.create(config);xxxxxxxxxx Config config = new Config();config.useSingleServer().setAddress("redis://myredisserver:6379");RedissonClient redisson = Redisson.create(config);// connects to 127.0.0.1:6379 by defaultRedissonClient redisson = Redisson.create();

SingleServerConfig singleConfig = config.useSingleServer();

SingleServerConfig类的设置参数如下

：

address（节点地址）

可以通过host:port的格式来指定节点地址。

subscriptionConnectionMinimumIdleSize（发布和订阅连接的最小空闲连接数）

默认值：1

用于发布和订阅连接的最小保持连接数（长连接）。Redisson内部经常通过发布和订阅来实现许多功能。长期保持一定数量的发布订阅连接是必须的。

subscriptionConnectionPoolSize（发布和订阅连接池大小）

默认值：50

用于发布和订阅连接的连接池最大容量。连接池的连接数量自动弹性伸缩。

connectionMinimumIdleSize（最小空闲连接数）

默认值：32

最小保持连接数（长连接）。长期保持一定数量的连接有利于提高瞬时写入反应速度。

connectionPoolSize（连接池大小）

默认值：64

连接池最大容量。连接池的连接数量自动弹性伸缩。

dnsMonitoring（是否启用DNS监测）

默认值：false

在启用该功能以后，Redisson将会监测DNS的变化情况。

dnsMonitoringInterval（DNS监测时间间隔，单位：毫秒）

默认值：5000

监测DNS的变化情况的时间间隔。

idleConnectionTimeout（连接空闲超时，单位：毫秒）

默认值：10000

如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值，那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。

connectTimeout（连接超时，单位：毫秒）

默认值：10000

同节点建立连接时的等待超时。时间单位是毫秒。

timeout（命令等待超时，单位：毫秒）

默认值：3000

等待节点回复命令的时间。该时间从命令发送成功时开始计时。

retryAttempts（命令失败重试次数）

默认值：3

如果尝试达到 retryAttempts（命令失败重试次数）仍然不能将命令发送至某个指定的节点时，将抛出错误。如果尝试在此限制之内发送成功，则开始启用 timeout（命令等待超时）计时。

retryInterval（命令重试发送时间间隔，单位：毫秒）

默认值：1500

在一条命令发送失败以后，等待重试发送的时间间隔。时间单位是毫秒。

reconnectionTimeout（重新连接时间间隔，单位：毫秒）

默认值：3000

当与某个节点的连接断开时，等待与其重新建立连接的时间间隔。时间单位是毫秒。

failedAttempts（执行失败最大次数）

默认值：3

在某个节点执行相同或不同命令时，连续失败 failedAttempts（执行失败最大次数）时，该节点将被从可用节点列表里清除，直到 reconnectionTimeout（重新连接时间间隔）超时以后再次尝试。

database（数据库编号）

默认值：0

尝试连接的数据库编号。

password（密码）

默认值：null

用于节点身份验证的密码。

subscriptionsPerConnection（单个连接最大订阅数量）

默认值：5

每个连接的最大订阅数量。

clientName（客户端名称）

默认值：null

在Redis节点里显示的客户端名称。

sslEnableEndpointIdentification（启用SSL终端识别）

默认值：true

开启SSL终端识别能力。

sslProvider（SSL实现方式）

默认值：JDK

确定采用哪种方式（JDK或OPENSSL）来实现SSL连接。

sslTruststore（SSL信任证书库路径）

默认值：null

指定SSL信任证书库的路径。

sslTruststorePassword（SSL信任证书库密码）

默认值：null

指定SSL信任证书库的密码。

sslKeystore（SSL钥匙库路径）

默认值：null

指定SSL钥匙库的路径。

sslKeystorePassword（SSL钥匙库密码）

默认值：null

指定SSL钥匙库的密码。

SpringBoot整合Redisson

Redisson有多种模式，首先介绍单机模式的整合。

一、导入Maven依赖

<!-- redisson-springboot -->
   <dependency>
       <groupId>org.redisson</groupId>
       <artifactId>redisson-spring-boot-starter</artifactId>
       <version>3.11.4</version>
   </dependency>

二、核心配置文件

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    database: 0
    timeout: 5000

三、添加配置类

RedissonConfig.java

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RedissonConfig {

    @Autowired
    private RedisProperties redisProperties;

    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        String redisUrl = String.format("redis://%s:%s", redisProperties.getHost() + "", redisProperties.getPort() + "");
        config.useSingleServer().setAddress(redisUrl).setPassword(redisProperties.getPassword());
        config.useSingleServer().setDatabase(3);
        return Redisson.create(config);
    }

}

自定义starter

由于redission可以有多种模式，处于学习的目的，将多种模式封装成一个start，可以学习一下starter的制作

在这里插入图片描述

封装一个RedissonManager，通过策略模式，根据不同的配置类型，创建 RedissionConfig实例，然后创建RedissionClient对象。

在这里插入图片描述

用RBucket操作分布式对象

Redission模拟了Java的面向对象编程思想，可以简单理解为一切皆为对象。

每一个 Redisson 对象实现了

RObject

and

RExpirable

两个interfaces

.

Usage example:

RObject object = redisson.get...()

object.sizeInMemory();

object.delete();

object.rename("newname");

object.isExists();

// catch expired event
object.addListener(new ExpiredObjectListener() {
   ...
});

// catch delete event
object.addListener(new DeletedObjectListener() {
   ...
});

每一个Redisson 对象的名字，就是 Redis中的 Key

.

Usage example:

RKeys keys = redisson.getKeys();

Iterable<String> allKeys = keys.getKeys();

Iterable<String> foundedKeys = keys.getKeysByPattern('key*');

long numOfDeletedKeys = keys.delete("obj1", "obj2", "obj3");

long deletedKeysAmount = keys.deleteByPattern("test?");

String randomKey = keys.randomKey();

long keysAmount = keys.count();

keys.flushall();

keys.flushdb();

Redisson通过RBucket接口代表可以访问任何类型的基础对象，或者普通对象

。

RBucket有一系列的工具方法，如compareAndSet()，get()，getAndDelete()，getAndSet()，set()，size()，trySet()等等，用于设值/取值/获取尺寸。

RBucket普通对象的最大大小，为512兆字节

。

RBucket<AnyObject> bucket = redisson.getBucket("anyObject");

bucket.set(new AnyObject(1));
AnyObject obj = bucket.get();

bucket.trySet(new AnyObject(3));
bucket.compareAndSet(new AnyObject(4), new AnyObject(5));
bucket.getAndSet(new AnyObject(6));

下面是一个完整的实例：

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

   @Test
    public void testRBucketExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient client = redissonManager.getRedisson();
        // RList 继承了 java.util.List 接口
        RBucket<String> rstring  = client.getBucket("redission:test:bucket:string");
        rstring.set("this is a string");

        RBucket<UserDTO> ruser  = client.getBucket("redission:test:bucket:user");
        UserDTO dto = new UserDTO();
        dto.setToken(UUID.randomUUID().toString());
        ruser.set(dto);
        System.out.println("string is: " + rstring.get());
        System.out.println("dto is: " + ruser.get());

        client.shutdown();
    }


}

运行上面的代码时，可以获得以下输出:

string is: this is a string
dto is: UserDTO(id=null, userId=null, username=null, password=null, nickname=null, token=183b6eeb-65a8-4b2a-80c6-cf17c08332ce, createTime=null, updateTime=null, headImgUrl=null, mobile=null, sex=null, enabled=null, type=null, openId=null, isDel=false)

在这里插入图片描述

使用 RList 操作 Redis 列表

下面的代码简单演示了如何在 Redisson 中使用
RList
对象。
RList
是 Java 的 List 集合的分布式并发实现。

考虑以下代码：

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

    @Test
    public void testListExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient client = redissonManager.getRedisson();
        // RList 继承了 java.util.List 接口
        RList<String> nameList = client.getList("redission:test:nameList");
        nameList.clear();
        nameList.add("张三");
        nameList.add("李四");
        nameList.add("王五");
        nameList.remove(-1);

        System.out.println("List size: " + nameList.size());


        boolean contains = nameList.contains("李四");
        System.out.println("Is list contains name '李四': " + contains);
        nameList.forEach(System.out::println);

        client.shutdown();
    }


}

使用 RList 操作 Redis 列表

下面的代码简单演示了如何在 Redisson 中使用
RList
对象。
RList
是 Java 的 List 集合的分布式并发实现。

考虑以下代码：

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

    @Test
    public void testListExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient client = redissonManager.getRedisson();
        // RList 继承了 java.util.List 接口
        RList<String> nameList = client.getList("redission:test:nameList");
        nameList.clear();
        nameList.add("张三");
        nameList.add("李四");
        nameList.add("王五");
        nameList.remove(-1);

        System.out.println("List size: " + nameList.size());


        boolean contains = nameList.contains("李四");
        System.out.println("Is list contains name '李四': " + contains);
        nameList.forEach(System.out::println);

        client.shutdown();
    }


}

运行上面的代码时，可以获得以下输出：

List size: 2
Is list contains name '李四': true
张三
李四

使用 RMap 操作 Redis 哈希

Redisson 还包括 RMap，它是 Java Map 集合的分布式并发实现，考虑以下代码：

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

    @Test
    public void testListExamples() {
         // 默认连接上 127.0.0.1:6379
        RedissonClient client = redissonManager.getRedisson();
        // RMap 继承了 java.util.concurrent.ConcurrentMap 接口
        RMap<String, Object> map = client.getMap("redission:test:personalMap");
        map.put("name", "张三");
        map.put("address", "北京");
        map.put("age", new Integer(50));

        System.out.println("Map size: " + map.size());

        boolean contains = map.containsKey("age");
        System.out.println("Is map contains key 'age': " + contains);
        String value = String.valueOf(map.get("name"));
        System.out.println("Value mapped by key 'name': " + value);

        client.shutdown();
    }


}

运行上面的代码时，将会看到以下输出：

Map size: 3
Is map contains key 'age': true
Value mapped by key 'name': 张三

在这里插入图片描述

执行 Lua脚本

Lua是一种开源、简单易学、轻量小巧的脚本语言，用标准C语言编写。

其设计的目的就是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。

Redis从2.6版本开始支持Lua脚本，Redis使用Lua可以：

原子操作。Redis会将整个脚本作为一个整体执行，不会被中断。可以用来批量更新、批量插入

减少网络开销。多个Redis操作合并为一个脚本，减少网络时延

代码复用。客户端发送的脚本可以存储在Redis中，其他客户端可以根据脚本的id调用。

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

    @Test
    public void testLuaExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient redisson = redissonManager.getRedisson();

        redisson.getBucket("redission:test:foo").set("bar");
        String r = redisson.getScript().eval(RScript.Mode.READ_ONLY,
                "return redis.call('get', 'redission:test:foo')", RScript.ReturnType.VALUE);
        System.out.println("foo: " + r);

        // 通过预存的脚本进行同样的操作
        RScript s = redisson.getScript();
        // 首先将脚本加载到Redis
        String sha1 = s.scriptLoad("return redis.call('get', 'redission:test:foo')");
        // 返回值 res == 282297a0228f48cd3fc6a55de6316f31422f5d17
        System.out.println("sha1: " + sha1);
        // 再通过SHA值调用脚本
        Future<Object> r1 = redisson.getScript().evalShaAsync(RScript.Mode.READ_ONLY,
                sha1,
                RScript.ReturnType.VALUE,
                Collections.emptyList());
        try {
            System.out.println("res: " + r1.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        client.shutdown();
    }
}

运行上面的代码时，将会看到以下输出：

foo: bar
sha1: 282297a0228f48cd3fc6a55de6316f31422f5d17
res: bar

在这里插入图片描述

使用 RLock 实现 Redis 分布式锁

RLock 是 Java 中可重入锁的分布式实现，下面的代码演示了 RLock 的用法：

public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

 @Test
    public void testLockExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient redisson = redissonManager.getRedisson();
        // RLock 继承了 java.util.concurrent.locks.Lock 接口
        RLock lock = redisson.getLock("redission:test:lock:1");

        final int[] count = {0};
        int threads = 10;
        ExecutorService pool = Executors.newFixedThreadPool(10);
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads; i++) {
            pool.submit(() ->
            {
                for (int j = 0; j < 1000; j++) {
                    lock.lock();

                    count[0]++;
                    lock.unlock();
                }
                countDownLatch.countDown();
            });
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("10个线程每个累加1000为： = " + count[0]);
        //输出统计结果
        float time = System.currentTimeMillis() - start;

        System.out.println("运行的时长为：" + time);
        System.out.println("每一次执行的时长为：" + time/count[0]);
    }

}

此代码将产生以下输出：

使用 RAtomicLong 实现 Redis 原子操作

RAtomicLong 是 Java 中 AtomicLong 类的分布式“替代品”，用于在并发环境中保存长值。以下示例代码演示了 RAtomicLong 的用法：


public class RedissionTest {

    @Resource
    RedissonManager redissonManager;

    @Test
    public void testRAtomicLongExamples() {
        // 默认连接上 127.0.0.1:6379
        RedissonClient redisson = redissonManager.getRedisson();
        RAtomicLong atomicLong = redisson.getAtomicLong("redission:test:myLong");
          // 线程数
        final int threads = 10;
        // 每条线程的执行轮数
        final int turns = 1000;
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        for (int i = 0; i < threads; i++)
        {
            pool.submit(() ->
            {
                try
                {
                    for (int j = 0; j < turns; j++)
                    {
                        atomicLong.incrementAndGet();
                    }

                } catch (Exception e)
                {
                    e.printStackTrace();
                }
            });
        }

        ThreadUtil.sleepSeconds(5);
        System.out.println("atomicLong: " + atomicLong.get());
        redisson.shutdown();
    }

}

此代码的输出将是：

atomicLong: 10000

简单Redision锁的原理

Redis发展到现在，几种常见的部署架构有：

单机模式；
哨兵模式；
集群模式；

先介绍，基于单机模式的简单Redision锁的使用。

简单Redision锁的使用

单机模式下，简单Redision锁的使用如下：

// 构造redisson实现分布式锁必要的Config
Config config = new Config();
config.useSingleServer().setAddress("redis://172.29.1.180:5379").setPassword("a123456").setDatabase(0);
// 构造RedissonClient
RedissonClient redissonClient = Redisson.create(config);
// 设置锁定资源名称
RLock disLock = redissonClient.getLock("DISLOCK");
//尝试获取分布式锁
boolean isLock= disLock.tryLock(500, 15000, TimeUnit.MILLISECONDS);
if (isLock) {
   try {
        //TODO if get lock success, do something;
        Thread.sleep(15000);

   } catch (Exception e) {
   } finally {
    // 无论如何, 最后都要解锁
    disLock.unlock();
   }
}

通过代码可知，经过Redisson的封装，实现Redis分布式锁非常方便，和显式锁的使用方法是一样的。RLock接口继承了 Lock接口。

我们再看一下Redis中的value是啥，和前文分析一样，hash结构, redis 的key就是资源名称。

hash结构的key就是UUID+threadId，hash结构的value就是重入值，在分布式锁时，这个值为1（Redisson还可以实现重入锁，那么这个值就取决于重入次数了）：

使用客户端工具看到的效果如下：

在这里插入图片描述

getLock()方法

可以看到，调用getLock()方法后实际返回一个RedissonLock对象

tryLock方法

下面来看下tryLock方法，源码如下：

    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
        
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

        try {
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
        
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

tryAcquire()方法

在RedissonLock对象的lock()方法主要调用tryAcquire()方法

tryLockInnerAsync

由于leaseTime == -1，于是走tryLockInnerAsync()方法，这个方法才是关键

首先，看一下evalWriteAsync方法的定义

我们可以知道，这里KEYS[1]就是getName()，ARGV[2]是getLockName(threadId)

假设：

前面获取锁时传的name是“DISLOCK”，
假设调用的线程ID是1，
假设成员变量UUID类型的id是01a6d806-d282-4715-9bec-f51b9aa98110
那么KEYS[1]=DISLOCK，ARGV[2]=01a6d806-d282-4715-9bec-f51b9aa98110:1
因此，这段脚本的意思是
1、判断有没有一个叫“DISLOCK”的key
2、如果没有，则在其下设置一个字段为“01a6d806-d282-4715-9bec-f51b9aa98110:1”，值为“1”的键值对 ，并设置它的过期时间
3、如果存在，则进一步判断“01a6d806-d282-4715-9bec-f51b9aa98110:1”是否存在，若存在，则其值加1，并重新设置过期时间
4、返回“DISLOCK”的生存时间（毫秒）

原理：加锁机制

这里用的数据结构是hash，hash的结构是： key 字段1 值1 字段2 值2 。。。

用在锁这个场景下，key就表示锁的名称，也可以理解为临界资源，字段就表示当前获得锁的线程

所有竞争这把锁的线程都要判断在这个key下有没有自己线程的字段，如果没有则不能获得锁，如果有，则相当于重入，字段值加1（次数）

在这里插入图片描述

Lua脚本的详解

为何要使用lua语言？

因为一大堆复杂的业务逻辑，可以通过封装在lua脚本中发送给redis，保证这段复杂业务逻辑执行的

原子性

在这里插入图片描述

关于 lua脚本的参数解释：

**KEYS[1]**代表的是你加锁的那个key，比如说：

RLock lock = redisson.getLock(“DISLOCK”);

这里你自己设置了加锁的那个锁key就是“DISLOCK”。

**ARGV[1]**代表的就是锁key的默认生存时间

调用的时候，传递的参数为 internalLockLeaseTime ，该值默认30秒。

**ARGV[2]**代表的是加锁的客户端的ID，类似于下面这样：

01a6d806-d282-4715-9bec-f51b9aa98110:1

lua脚本的第一段if判断语句，就是用“exists DISLOCK”命令判断一下，如果你要加锁的那个锁key不存在的话，你就进行加锁。

如何加锁呢？很简单，用下面的redis命令：

hset DISLOCK 01a6d806-d282-4715-9bec-f51b9aa98110:1 1

通过这个命令设置一个hash数据结构，这行命令执行后，会出现一个类似下面的数据结构：

DISLOCK:
    {
        8743c9c0-0795-4907-87fd-6c719a6b4586:1 1
    }

接着会执行“pexpire DISLOCK 30000”命令，设置DISLOCK这个锁key的生存时间是30秒(默认)

锁互斥机制

那么在这个时候，如果客户端2来尝试加锁，执行了同样的一段lua脚本，会咋样呢？

很简单，第一个if判断会执行“exists DISLOCK”，发现DISLOCK 这个锁key已经存在了。

接着第二个if判断，判断一下，DISLOCK锁key的hash数据结构中，是否包含客户端2的ID，但是明显不是的，因为那里包含的是客户端1的ID。

所以，客户端2会获取到pttl DISLOCK返回的一个数字，这个数字代表了DISLOCK 这个锁key的**剩余生存时间。**比如还剩15000毫秒的生存时间。

此时客户端2会进入一个while循环，不停的尝试加锁。

可重入加锁机制

如果客户端1都已经持有了这把锁了，结果可重入的加锁会怎么样呢？

RLock lock = redisson.getLock(“DISLOCK”)

lock.lock();

//业务代码

lock.lock();

//业务代码

lock.unlock();

lock.unlock()

分析上面那段lua脚本。

第一个if判断肯定不成立，“exists DISLOCK”会显示锁key已经存在了。

第二个if判断会成立，因为DISLOCK的hash数据结构中包含的那个ID，就是客户端1的那个ID，也就是“8743c9c0-0795-4907-87fd-6c719a6b4586:1”

此时就会执行可重入加锁的逻辑，他会用：

incrby DISLOCK

8743c9c0-0795-4907-87fd-6c719a6b4586:1 1

通过这个命令，对客户端1的加锁次数，累加1。

此时DISLOCK数据结构变为下面这样：

DISLOCK:
    {
        8743c9c0-0795-4907-87fd-6c719a6b4586:1 2
    }

释放锁机制

如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑也是非常简单的。

其实说白了，就是每次都对DISLOCK数据结构中的那个加锁次数减1。

如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用：

“del DISLOCK”命令，从redis里删除这个key。

然后呢，另外的客户端2就可以尝试完成加锁了。

unlock 源码

  @Override
    public void unlock() {
        try {
            get(unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException e) {
            if (e.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException) e.getCause();
            } else {
                throw e;
            }
        }
        
//        Future<Void> future = unlockAsync();
//        future.awaitUninterruptibly();
//        if (future.isSuccess()) {
//            return;
//        }
//        if (future.cause() instanceof IllegalMonitorStateException) {
//            throw (IllegalMonitorStateException)future.cause();
//        }
//        throw commandExecutor.convertException(future);
    }

我们还是假设name=DISLOCK，假设线程ID是1

同理，我们可以知道

KEYS[1]是getName()，即KEYS[1]=DISLOCK

KEYS[2]是getChannelName()，即KEYS[2]=redisson_lock__channel:{DISLOCK}

ARGV[1]是LockPubSub.unlockMessage，即ARGV[1]=0

ARGV[2]是生存时间

ARGV[3]是getLockName(threadId)，即ARGV[3]=8743c9c0-0795-4907-87fd-6c719a6b4586:1

因此，上面脚本的意思是：

1、判断是否存在一个叫“DISLOCK”的key

2、如果不存在，返回nil

3、如果存在，使用Redis Hincrby 命令用于为哈希表中的字段值加上指定增量值 -1 ，代表减去1

4、若counter >，返回空，若字段存在，则字段值减1

5、若减完以后，counter > 0 值仍大于0，则返回0

6、减完后，若字段值小于或等于0，则用 publish 命令广播一条消息，广播内容是0，并返回1；

可以猜测，广播0表示资源可用，即通知那些等待获取锁的线程现在可以获得锁了

在这里插入图片描述

通过redis Channel 解锁订阅

以上是正常情况下获取到锁的情况，那么当无法立即获取到锁的时候怎么办呢？

再回到前面获取锁的位置

@Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

    //    订阅
    RFuture<RedissonLockEntry> future = subscribe(threadId);
    commandExecutor.syncSubscription(future);

    try {
        while (true) {
            ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }

            // waiting for message
            if (ttl >= 0) {
                getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                getEntry(threadId).getLatch().acquire();
            }
        }
    } finally {
        unsubscribe(future, threadId);
    }
//        get(lockAsync(leaseTime, unit));
}


protected static final LockPubSub PUBSUB = new LockPubSub();

protected RFuture<RedissonLockEntry> subscribe(long threadId) {
    return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}

protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {
    PUBSUB.unsubscribe(future.getNow(), getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}

这里会订阅Channel，当资源可用时可以及时知道，并抢占，防止无效的轮询而浪费资源

这里的channel为

redisson_lock__channel:{DISLOCK}

在这里插入图片描述

当资源可用用的时候，循环去尝试获取锁，由于多个线程同时去竞争资源，所以这里用了信号量，对于同一个资源只允许一个线程获得锁，其它的线程阻塞

watch dog自动延期机制

客户端1加锁的锁key默认生存时间才30秒，如果超过了30秒，客户端1还想一直持有这把锁，怎么办呢？

使用watchDog机制实现锁的续期

但是聪明的同学肯定会问：

有效时间设置多长，假如我的业务操作比有效时间长，我的业务代码还没执行完，就自动给我解锁了，不就完蛋了吗。

这个问题就有点棘手了，在网上也有很多讨论：

第一种解决方法就是靠程序员自己去把握，预估一下业务代码需要执行的时间，然后设置有效期时间比执行时间长一些，保证不会因为自动解锁影响到客户端业务代码的执行。

但是这并不是万全之策，比如网络抖动这种情况是无法预测的，也有可能导致业务代码执行的时间变长，所以并不安全。

第二种方法，使用监事狗watchDog机制实现锁的续期。

第二种方法比较靠谱一点，而且无业务入侵。

在Redisson框架实现分布式锁的思路，就使用watchDog机制实现锁的续期。

当加锁成功后，同时开启守护线程，默认有效期是30秒，每隔10秒就会给锁续期到30秒，只要持有锁的客户端没有宕机，就能保证一直持有锁，直到业务代码执行完毕由客户端自己解锁，如果宕机了自然就在有效期失效后自动解锁。

这里，和前面解决 JVM STW的锁过期问题有点类似，只不过，watchDog自动续期，也没有完全解决JVM STW的锁过期问题。

如何彻底解决 JVM STW的锁过期问题，可以来疯狂创客圈的社群讨论。

Redisson watchdog 使用和原理

实际上，redisson加锁的基本流程图如下：

在这里插入图片描述

这里专注于介绍watchdog。

首先watchdog的具体思路是加锁时，默认加锁 30秒，每10秒钟检查一次，如果存在就重新设置过期时间为30秒。

然后设置默认加锁时间的参数是 lockWatchdogTimeout（监控锁的看门狗超时，单位：毫秒）

官方文档描述如下

lockWatchdogTimeout（监控锁的看门狗超时，单位：毫秒）

默认值：30000

监控锁的看门狗超时时间单位为毫秒。该参数只适用于分布式锁的加锁请求中未明确使用leaseTimeout参数的情况。如果该看门狗未使用lockWatchdogTimeout去重新调整一个分布式锁的lockWatchdogTimeout超时，那么这个锁将变为失效状态。这个参数可以用来避免由Redisson客户端节点宕机或其他原因造成死锁的情况。

需要注意的是

1.watchDog 只有在未显示指定加锁时间时才会生效。（这点很重要）

2.lockWatchdogTimeout设定的时间不要太小，比如我之前设置的是 100毫秒，由于网络直接导致加锁完后，watchdog去延期时，这个key在redis中已经被删除了。

watchLog总结

1.要使 watchLog机制生效，lock时不要设置过期时间

2.watchlog的延时时间可以由 lockWatchdogTimeout指定默认延时时间，但是不要设置太小。如100

3.watchdog 会每 lockWatchdogTimeout/3时间，去延时。

4.watchdog 通过类似netty的 Future功能来实现异步延时

5.watchdog 最终还是通过 lua脚本来进行延时

Redisson框架的分布式锁

Redisson框架十分强大，除了前面介绍的 getLock方法获取的分布式锁（输入可重入锁的类型），还有很多其他的分布式锁类型。

总体的Redisson框架的分布式锁类型，大致如下：

可重入锁
公平锁
联锁
红锁
读写锁
信号量
可过期信号量
闭锁（/倒数闩）

未完待续。。。。

数据库锁

数据库乐观锁

Redis实现思路

手工造轮子：基于Jedis 的API实现分布式锁

Redis几种架构

基于Jedis API的分布式锁的总体流程：

简单加锁：使用set的命令时，同时设置过期时间

基于Jedis 的API实现简单解锁代码

再造轮子：基于Lua脚本实现分布式锁

lua脚本的好处

基于纯Lua脚本的分布式锁的执行流程

加锁的Lua脚本： lock.lua

解锁的Lua脚本： unlock.lua：

两个文件，放在资源文件夹下备用：

在Java中调用lua脚本，完成加锁操作

在Java中调用lua脚本，完成解锁操作

编写RedisLockService用于管理JedisLock

测试用例

STW导致的锁过期问题

Redisson简介

Redisson与Jedis的对比

1.概况对比

2.可伸缩性

3.第三方框架整合

Redission 的源码地址：

特性 & 功能：

Redisson的使用

如何安装 Redisson

获取RedissonClient对象

SpringBoot整合Redisson

一、导入Maven依赖

二、核心配置文件

三、添加配置类

用RBucket操作分布式对象

使用 RList 操作 Redis 列表

使用 RList 操作 Redis 列表

使用 RMap 操作 Redis 哈希

执行 Lua脚本

使用 RLock 实现 Redis 分布式锁

使用 RAtomicLong 实现 Redis 原子操作

简单Redision锁的原理

简单Redision锁的使用

getLock()方法

tryLock方法

tryAcquire()方法

原理：加锁机制

Lua脚本的详解

锁互斥机制

可重入加锁机制

RLock lock = redisson.getLock(“DISLOCK”) lock.lock(); //业务代码 lock.lock(); //业务代码 lock.unlock(); lock.unlock()

释放锁机制

unlock 源码

通过redis Channel 解锁订阅

watch dog自动延期机制

使用watchDog机制实现锁的续期

Redisson watchdog 使用和原理

Redisson框架的分布式锁

你可能也喜欢

RLock lock = redisson.getLock(“DISLOCK”)

lock.lock();

//业务代码

lock.lock();

//业务代码

lock.unlock();

lock.unlock()