用万字长文来讲讲本地锁至分布式锁的演进和Redis实现，扩展 R

作者: 程序花生 | 来源:发表于2022-10-21 11:09 被阅读0次

本文就是针对上一篇讨论的数据一致性中的加锁方案来进行一个从浅至深的探究。

文章大纲：

本文字数约为1w字左右，和上一篇差不多的字数，建议可以腾出一点点空闲时间来阅读，都是我的存稿~ 哈哈，寻思着再不发，这个月都得过完了。

关于代码：

后期在检查时，发现案例中的代码是有点不太合适的，应当将所有案例中的递归调用方法改为循环重试，并限制重试次数，而非一直递归调用。切记切记切记👨💻。

一、本地锁

1.1、本地锁的使用

本地锁主要是针对单体服务而言的，锁的都是单体应用内的进程。

像之前在单机情况下出现的读写并发情况。因为并发情况下网络出现问题或是出现其他卡顿问题，导致执行顺序发生变化，从而产生了数据不一致性。如下图：

解决并发最快的方式就是加锁吗，我们也就给它来把锁吧，Java中的锁是有蛮多的，我这里不过多讨论啦（synchronized、JUC）等等。

我案例中所使用的是 JUC 包下的读写锁ReentrantReadWriteLock ，毕竟不能让锁直接限制了Redis 的发挥~，读写锁是读并发，写独占的模式。

增加读写锁之后的流程图如下：

图片说明：加上锁之后的流程

案例代码如下：

/**
 * @description: 单机redis下的操作
 * @author: Ning Zaichun
 * @date: 2022年09月06日 23:20
 */
@Slf4j
@Service
public class RedisCacheServiceImpl implements IRedisCacheService {

    @Autowired
    private MenuMapper menuMapper;

    @Autowired
    StringRedisTemplate redisTemplate;

    private static final String REDIS_MENU_CACHE_KEY = "menu:list";

    private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    @Override
    public List<MenuEntity> getList() {
        //1、从缓存中获取
        String menuJson = redisTemplate.opsForValue().get(REDIS_MENU_CACHE_KEY);
        if (!StringUtils.isEmpty(menuJson)) {
            System.out.println("缓存中有，直接从缓存中获取");
            //2、缓存不为空直接返回
            List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
            });
            return menuEntityList;
        }
        //3、查询数据库
        //不加锁情况下
        // List<MenuEntity> noLockList = getMenuJsonFormDb();
        // 加锁情况下
        List<MenuEntity> menuEntityList = getMenuJsonFormDbWithReentrantReadWriteLock();
        return menuEntityList;
    }

    public List<MenuEntity> getMenuJsonFormDb() {
        System.out.println("缓存中没有，重新从数据中查询~==>");
        //缓存为空，查询数据库，重新构建缓存
        List<MenuEntity> result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
        //4、将查询的结果，重新放入缓存中
        redisTemplate.opsForValue().set(REDIS_MENU_CACHE_KEY, JSON.toJSONString(result));
        return result;
    }

    public List<MenuEntity> getMenuJsonFormDbWithReentrantReadWriteLock() {
        List<MenuEntity> result = null;
        System.out.println("缓存中没有，加锁，重新从数据中查询~==>");
        // synchronized 是同步锁，所以当多个线程同时执行到这里时，会阻塞式等待
        ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();
        readLock.lock();
        try {
            String menuJson = redisTemplate.opsForValue().get(REDIS_MENU_CACHE_KEY);
            //加锁成功... 再次判断缓存是否为空
            if (!StringUtils.isEmpty(menuJson)) {
                System.out.println("缓存中，直接从缓存中获取");
                //2、缓存不为空直接返回
                List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
                });
                return menuEntityList;
            }
            //缓存为空，查询数据库，重新构建缓存
            result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
            //4、将查询的结果，重新放入缓存中
            redisTemplate.opsForValue().set(REDIS_MENU_CACHE_KEY, JSON.toJSONString(result));
            return result;
        } finally {
            readLock.unlock();
        }
    }

    @Override
    public Boolean updateMenuById(MenuEntity menu) {
        //return updateMenuByIdNoWithLock(menu);
        return updateMenuByIdWithReentrantReadWriteLock(menu);
    }

    public Boolean updateMenuByIdNoWithLock(MenuEntity menu) {
        // 1、删除缓存
        redisTemplate.delete(REDIS_MENU_CACHE_KEY);
        System.out.println("清空单机Redis缓存");
        // 2、更新数据库
        return menuMapper.updateById(menu) > 0;
    }

    public Boolean updateMenuByIdWithReentrantReadWriteLock(MenuEntity menu) {
        ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
        writeLock.lock();
        try {
            // 1、删除缓存
            System.out.println("清除缓存");
            redisTemplate.delete(REDIS_MENU_CACHE_KEY);
            // 2、更新数据库
            return menuMapper.updateById(menu) > 0;
        }finally {
            writeLock.unlock();
        }
    }
}

具体还需大家去了解~

我这里加的 JUC 下的读写锁，原本我想的是弄的JUC的锁

1.2、本地锁存在的问题

看起来本地锁没有并发问题，不管有多少请求一起进来，都要去争取那唯一的一把锁，抢到了才能继续往下执行业务。

单体项目中，每把锁锁的就是当前服务中的当前线程的请求。

图片说明：单体服务时

但是当服务需要进一步扩展时，就会随之产生出一些问题。

多服务并发时，如果还是只给当前线程加锁，多个用户一起尝试获取锁时，可能会有多个用户同时获取到锁，导致出现问题。

如下图：

每个服务都是单独的，加锁操作也只是给自己的，大家不能共享，那么实际上在高并发的时候，是根本没效果的。

我1号服务抢到了锁，还没等到释放，2号服务又获取到了锁，接着3号、4号等等，大家都可以操作数据库，这把锁也就失去了它该有的作用。

因此就进一步出现了分布式锁，接下来继续看吧。

二、分布式锁的介绍

本地锁失效是因为无法锁住各个应用的读写请求，失效的根本原因就是其他的服务无法感知到是否已经有请求上锁了，即无法共享锁信息。

分布式锁，其实也就是将加锁的这一个操作，单独的抽取出来了，让每个服务都能感知到。

之前就说了，软件架构设计中，"没有什么是加一层解决不了的，如果加一层不行就再加一层"。

这里其实也是一样，只不过碰巧这一层可以在Redis中实现罢了，看起来倒是没有多加一层，但如果是用Zookeeper 或者其他方式来实现，你会发现架构中会多一层滴。

其实理解思想实现的方式有很多种的，

Redis 实现分布式锁
Zookeeper 实现分布式锁
MySQL 专门用一张表来记录信息，实现分布式锁，也是常说的基于数据库实现分布式锁。

所谓的加锁，其本质也就是判断一个信号量是否存在罢了，分布式也就是把这个信号量从本地线程中，移植到了Redis中存储，让所有服务中的请求都能共享一把锁。知道思想后，实现方式并不局限，大家也不要局限了自己，都已经站在巨人肩膀上，就要想的更多一些~

我采用 Redis 实现分布式锁，主要原因：

Redis 是基于内存操作的数据库，速度快；
市场主流的数据库，拥有较多的参考资料；
Redis 社区开发者活跃，并且 Redis 对分布式锁有较好的支持；

今天所讨论的，主要就是针对于使用 Redis 实现分布式锁，流程图大致如下：

图片说明：此图为获取锁的大致流程，其之后的构建缓存、释放锁等未在图上所标明

虽然两个服务都是独立的，但是在执行数据库代码前，都需要先获取到读锁或者写锁，以确保并发时执行的正确性~

接下来就是说分布式锁的实现啦~

三、分布式锁的实现

在上一小节就已说分布式锁的实现有多种方式，大的范围中有 Redis、Zookeeper、MySQL等实现方式，我具体讲的是以 Redis 的实现。

讲解过程也是逐步深入，逐步演进，并非是直接丢出实现代码，针对为什么要这么做，为什么最终是这样，让大家有一个了解过程。

锁的第一个要求就是要能做到互斥，而在Redis中最容易想到，也是最简单的，无疑就是 setnx 命令。

我们就以 setnx抛砖引玉，来对分布式锁的实现，做一个逐步演进的讨论。

3.1、setnx

Redis Setnx（ SET IF Not EXists ）命令在指定的 key 不存在时，为 key 设置指定的值，这种情况下等同 SET 命令。当 key存在时，什么也不做。

返回值

如果key设置成功了，则返回1
如果key设置失败了，则返回0

redis> SETNX mykey "Hello"
(integer) 1
redis> SETNX mykey "World"
(integer) 0
redis> GET mykey
"Hello"

转换到加锁的流程中就是，当读请求过来的时候，先用setnx命令往 Redis 中设置一个值，返回1则是加锁成功了，返回0则是加锁失败了。

如何释放锁呢？

setnx其本质和set命令其实差不多，都是往 Redis 中设置一个值，释放锁，也就是删除这个key值，使用 del key 命令删除即等于释放锁~

思路👨🏫：

读请求过来，判断缓存中是否存在数据，存在立马返回，不存在则往下走。
读请求尝试获取读锁，使用 setnx 命令
向Redis中尝试set一个值，当且仅当值不存在时设置成功。
设置成功返回 1，否则返回 0
获取成功后，查询数据库，重新构建缓存。

代码案例实现：

/**
 * @description:
 * @author: Ning Zaichun
 * @date: 2022年09月20日 20:59
 */
@Service
public class SetNxExLockServiceImpl implements ISetNxExLockService {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MenuMapper menuMapper;

    private static final String SET_NX_EX_MENU_LIST = "set:nx:ex:menu:list";
    private static final String SET_NX_EX_MENU_LIST_LOCK = "set:nx:ex:menu:list:lock";
    private static final String SET_NX_EX_MENU_LIST_LOCK_VALUE = "lock";

    @Override
    public List<MenuEntity> getList() {
        // 判断缓存是否有数据
        String menuJson = stringRedisTemplate.opsForValue().get(SET_NX_EX_MENU_LIST);
        if (menuJson != null) {
            System.out.println("缓存中有，直接返回缓存中的数据");
            List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
            });
            return menuEntityList;
        }

        // 从数据库中查询
        List<MenuEntity> result = getMenuJsonFormDbWithLock();

        return result;
    }

    /**
     * 问题：死锁问题， 如果在执行解锁操作之前，服务突然宕机，那么锁就永远无法被释放，从而造成死锁问题
     * 解决方案: 因为 Redis 的更新，现在 setIfAbsent 支持同时设置过期时间，而无需分成两步操作。
     *
     * @return
     */
    public List<MenuEntity> getMenuJsonFormDbWithLock() {
        List<MenuEntity> result = null;
        System.out.println("缓存中没有，加锁，重新从数据中查询~==>");
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, SET_NX_EX_MENU_LIST_LOCK_VALUE);
        if (lock) {
            System.out.println("获取分布式锁成功...");
            try {
                //加锁成功...执行业务
                result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
                stringRedisTemplate.opsForValue().set(SET_NX_EX_MENU_LIST, JSON.toJSONString(result));
            } finally {
                // 释放锁
                stringRedisTemplate.delete(SET_NX_EX_MENU_LIST_LOCK);
            }
            return result;
        } else {
            System.out.println("获取分布式锁失败...等待重试...");
            //加锁失败...重试机制
            //休眠一百毫秒
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 手动实现自旋
            return getMenuJsonFormDbWithLock();
        }
    }

        // 更新操作
        @Override
        public Boolean updateMenuById(MenuEntity menu) {
            return updateMenuByIdWithLock(menu);
        }
        public Boolean updateMenuByIdWithLock(MenuEntity menu) {
            Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, SET_NX_EX_MENU_LIST_LOCK_VALUE);
            Boolean update = false;
            if (lock) {
                System.out.println("更新操作：获取分布式锁成功===>");
                // 删除缓存
                try {
                    stringRedisTemplate.delete(SET_NX_EX_MENU_LIST);
                    // 更新数据库
                    update = menuMapper.updateById(menu) > 0;
                } finally {
                    // 一定要释放锁，以免造成死锁问题
                    stringRedisTemplate.delete(SET_NX_EX_MENU_LIST_LOCK);
                }
                return update;
            }else{
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                return updateMenuByIdWithLock(menu);
            }
        }

}

注意：代码中有加锁操作，就一定要记得将解锁操作放到了finally执行，以保证在代码层面不会出现死锁问题。

你觉得上面的案例存在什么样的问题吗？

咋一看，上面的好像确实实现了分布式锁，setIfAbsent()方法实现了锁的互斥性，Redis 也让锁得到共享，但是真的没问题吗？

思考：

如果某个时刻代码执行到释放锁的那一步，服务突然挂了，是不是也就意味着锁释放失败了勒，那么也就造成了最大的问题死锁，因为没有解锁操作，锁也就将无法被释放。

那该如何解决呢？

最佳的方式就是给锁设置一个过期时间~

3.2、解决死锁问题

最容易想到也是最简单的方式就是给这个分布式锁加个过期时间，比如3s、5s之类，使用EXPIRE设置一个过期时间

但如果想到的是这种：

redis> SETNX mykey "Hello"
(integer) 1
redis> EXPIRE mykey 10
(integer) 1
redis> SETNX mykey "World"
(integer) 0
redis> TTL mykey
(integer) 8
redis> GET mykey
"Hello"

如果是这么实现的话，咱们就还是对于 Redis 了解太少了。

首先说说这样操作会出现的问题：

我成功设置了key，但是还没执行到设置时间那一步，应用程序突然挂了，导致死锁问题产生。
或者是成功设置了 key 后，Redis 服务突然崩溃了，不干活了，这也导致了后面的EXPIRE无法执行成功，同样会产生死锁问题。

【重点】：因为加锁和设置时间不是一个原子性操作

怎么才能将加锁和设置锁时间变成一个原子操作呢？

其实这一步Redis已经帮我们做好啦~

Redis 中的set命令是可以添加参数的，一条set命令即可实现SETNX+EXPIRE效果，将加锁和设置时间变成一个原子性操作，要么一起成功，要么一起失败。

完整命令参数：文档链接

SET key value [EX seconds|PX milliseconds|KEEPTTL] [NX|XX] [GET]

EX seconds – 设置键key的过期时间，单位时秒
PX milliseconds – 设置键key的过期时间，单位时毫秒
NX – 只有键key不存在的时候才会设置key的值
XX – 只有键key存在的时候才会设置key的值
KEEPTTL -- 获取 key 的过期时间
GET -- 返回 key 存储的值，如果 key 不存在返回空

注意: 由于SET命令加上选项已经可以完全取代SETNX, SETEX, PSETEX, GETSET,的功能，所以在将来的版本中，redis可能会不推荐使用并且最终抛弃这几个命令。

返回值

字符串: 如果SET命令正常执行那么回返回OK 多行字符串: 使用 GET 选项，返回 key 存储的值，如果 key 不存在返回空: 否则如果加了NX 或者 XX选项，SET 没执行，那么会返回nil。

例子：

从这个小案例中可以看出，这是符合我们要的命令的~

Java 代码实现：这一步的代码实现和上一小节相比，仅改动了一行代码：

上一小节：

Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, SET_NX_EX_MENU_LIST_LOCK_VALUE);

加上过期时间：

Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, SET_NX_EX_MENU_LIST_LOCK_VALUE, 5L, TimeUnit.SECONDS);

就这样，再那样，再这样，你看死锁问题就被解决啦😜

死锁问题确实被解决了，现在你觉得还有问题吗？

我们把它的每一个步骤都拆分来看，就会出现下面的这样一个场景：

假设我们约定锁过期时间为5s，但是执行这个业务时突然卡起来，此业务执行时间超过了我们预估的 5s，那么就可能出现以下情况：

第一条线程抢到锁，业务执行超时，第一条线程所持有的锁被自动释放；此时第二条线程拿到锁，准备执行业务，刚好第一条线程业务执行完成，照常执行了释放锁的过程，导致第二条线程持有的锁被第一条线程所释放，锁被其他人释放。

这个情况中存在两个问题：

业务执行超时，锁被自动释放
锁被其他人释放，导致业务出现问题

关于第一个问题，就是常说的锁续期问题，这点之后在使用 Redission 时再细谈。

第二个问题，就比较好解决了，我们每次加锁的时候，带上自己的身份标识，在解锁的时候，进行一次判断即可。

接着往下看吧 👇

3.3、锁被其他人释放，该怎么办？

我们每次加锁的时候，带上自己的身份标识，在解锁的时候，进行一次判断即可。

比如：加锁的时候，生成一个UUID作为 KEY，释放锁时，获取一下锁，判断一下UUID是否相等，相等则执行删除，否则不执行删除。

在 Redis 中命令演示如下：

设置锁：set uuid "lock" NX EX 60

释放锁：
1、get uuid，身份标识相等，则执行2；
2、 del uuid

Java 代码如下：

/**
 * @description:
 * @author: Ning Zaichun
 * @date: 2022年09月20日 20:59
 */
@Service
public class SetNxExLockServiceImpl implements ISetNxExLockService {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MenuMapper menuMapper;

    private static final String SET_NX_EX_MENU_LIST = "set:nx:ex:menu:list";
    private static final String SET_NX_EX_MENU_LIST_LOCK = "set:nx:ex:menu:list:lock";
    private static final String SET_NX_EX_MENU_LIST_LOCK_VALUE = "lock";

    @Override
    public List<MenuEntity> getList() {
        // 判断缓存是否有数据
        String menuJson = stringRedisTemplate.opsForValue().get(SET_NX_EX_MENU_LIST);
        if (menuJson != null) {
            System.out.println("缓存中有，直接返回缓存中的数据");
            List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
            });
            return menuEntityList;
        }

        // 从数据库中查询
        List<MenuEntity> result = getMenuJsonFromDbWithRedisLock();

        return result;
    }

    /**
     * 问题：锁被其他人释放，这该如何处理。
     * 加锁的时候，将值给定位一个唯一的标识符（我这里使用的是 UUID ）
     * 1、解锁之前，先判断是不是自己获取的那把锁，
     * 2、确定是一把锁就执行 解锁锁操作
     *
     * @return
     */
    public List<MenuEntity> getMenuJsonFromDbWithRedisLock() {
        List<MenuEntity> result = new ArrayList<>();
        System.out.println("缓存中没有，加锁，重新从数据中查询~==>");
        // 给锁设定一个时间
        String uuid = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, uuid, 5L, TimeUnit.SECONDS);
        if (lock) {
            System.out.println("获取分布式锁成功...");
            try {
                //加锁成功...执行业务
                result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
                stringRedisTemplate.opsForValue().set(SET_NX_EX_MENU_LIST, JSON.toJSONString(result));
            } finally {
                // 获取锁，判断是不是当前线程的锁
                String token = stringRedisTemplate.opsForValue().get(SET_NX_EX_MENU_LIST_LOCK);
                if (uuid.equals(token)) {
                    // 确定是同一把锁， 才释放锁
                    stringRedisTemplate.delete(SET_NX_EX_MENU_LIST_LOCK);
                }
            }
            return result;
            /**
             * 那这样就没有问题了吗？
             * 并不是。
             * 这里存在的问题也很明显，删除操作已经不在是一个原子性操作了。
             * 1、一个是查询判断
             * 2、第二个才是解锁操作
             * 那么又会拆分成，如果我第一步执行成功，第二步执行失败的场景，所以我们要把它变成原子操作才行。
             */
        } else {
            System.out.println("获取分布式锁失败...等待重试...");
            //加锁失败...重试机制
            //休眠一百毫秒
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return getMenuJsonFromDbWithRedisLock();
        }

    }
    @Override
    public Boolean updateMenuById(MenuEntity menu) {
        //        return updateMenuByIdWithLock(menu);
        //        return updateMenuWithExpireLock(menu);
        return updateMenuWithRedisLock(menu);
    }

    public Boolean updateMenuWithExpireLock(MenuEntity menu) {
        // 给锁设定一个时间
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, SET_NX_EX_MENU_LIST_LOCK_VALUE, 5L, TimeUnit.SECONDS);
        Boolean update = false;
        if (lock) {
            System.out.println("更新操作：获取分布式锁成功===> 清除缓存");
            try {
                // 删除缓存
                stringRedisTemplate.delete(SET_NX_EX_MENU_LIST);
                // 更新数据库
                update = menuMapper.updateById(menu) > 0;
            } finally {
                // 一定要释放锁，以免造成死锁问题
                stringRedisTemplate.delete(SET_NX_EX_MENU_LIST_LOCK);
            }
            return update;
        }  else{
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return updateMenuWithExpireLock(menu);
        }
    }

}

那么这就没有问题了吗？

并不是。这里存在的问题也很明显，删除操作又已经不再是一个原子性操作了。

第一步是查询判断
第二步才是解锁操作

那么继而又会出现，我第一步执行成功，第二步执行失败的场景，所以我们必须要把它变成原子性操作才行。

这个时就要用到了 Redis 中的 lua 脚本啦~

让它去帮助我们实现将判断和解锁变成一步原子性操作~ 接着看吧

3.4、lua 脚本实现分布式锁

其实有阅读过 Redis 官方文档的朋友，在看上面的那个 set命令的文档时，就会发现，其实滑到下半部分，Redis 就有提到不推荐使用SET resource-name anystring NX EX max-lock-time 这个简单方法来实现分布式锁~

并且也给出了相应的建议：如下图

if redis.call("get",KEYS[1]) == ARGV[1]
then
    return redis.call("del",KEYS[1])
else
    return 0
end

我们使用的其实就是上面的脚本，哈哈

/**
 * @description:
 * @author: Ning Zaichun
 * @date: 2022年09月20日 20:59
 */
@Service
public class SetNxExLockServiceImpl implements ISetNxExLockService {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MenuMapper menuMapper;

    private static final String SET_NX_EX_MENU_LIST = "set:nx:ex:menu:list";
    private static final String SET_NX_EX_MENU_LIST_LOCK = "set:nx:ex:menu:list:lock";
    private static final String SET_NX_EX_MENU_LIST_LOCK_VALUE = "lock";

    @Override
    public List<MenuEntity> getList() {
        // 判断缓存是否有数据
        String menuJson = stringRedisTemplate.opsForValue().get(SET_NX_EX_MENU_LIST);
        if (menuJson != null) {
            System.out.println("缓存中有，直接返回缓存中的数据");
            List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
            });
            return menuEntityList;
        }

        // 从数据库中查询
        List<MenuEntity> result = getMenuJsonFormDbWithLock();

        return result;
    }

    /**
     * 问题：释放锁操作 丢失 原子性
     * 解决方案: lua 脚本
     * 既然删除操作变成了两步，失去了原子性，那么我们就把它改成一步就行啦呀，
     * 此时就得用上我们的 lua 脚本了
     *
     * @return
     */
    public List<MenuEntity> getMenuJsonFormDbWithLuaLock() {
        List<MenuEntity> result = new ArrayList<>();
        System.out.println("缓存中没有，加锁，重新从数据中查询~==>");
        // 给锁设定一个时间
        String uuid = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, uuid, 5L, TimeUnit.SECONDS);
        if (lock) {
            System.out.println("获取分布式锁成功...");
            try {
                //加锁成功...执行业务
                result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
                stringRedisTemplate.opsForValue().set(SET_NX_EX_MENU_LIST, JSON.toJSONString(result));
            } finally {
                // 编写 lua 脚本
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                //删除锁  execute 这个放番中的参数可能还需要提一提
                stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Arrays.asList(SET_NX_EX_MENU_LIST_LOCK), uuid);
            }
            return result;
        } else {
            System.out.println("获取分布式锁失败...等待重试...");
            //加锁失败...重试机制
            //休眠一百毫秒
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return getMenuJsonFormDbWithLuaLock();
        }
    }

    @Override
    public Boolean updateMenuById(MenuEntity menu) {
//        return updateMenuByIdWithLock(menu);
//        return updateMenuWithExpireLock(menu);
        return updateMenuWithLuaLock(menu);
    }
    /**
     * 问题：释放锁操作 丢失 原子性
     * 解决方案: lua 脚本
     * 既然删除操作变成了两步，失去了原子性，那么我们就把它改成一步就行啦呀，
     * 此时就得用上我们的 lua 脚本了
     *
     * @return
     */
    public Boolean updateMenuWithRedisLock(MenuEntity menu) {
        String uuid = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent(SET_NX_EX_MENU_LIST_LOCK, uuid, 5L, TimeUnit.SECONDS);
        Boolean update = false;
        if (lock) {
            System.out.println("获取分布式锁成功...");
            try {
                //加锁成功...执行业务
                stringRedisTemplate.delete(SET_NX_EX_MENU_LIST);
                // 更新数据库
                update = menuMapper.updateById(menu) > 0;
            } finally {
                // 获取锁，判断是不是当前线程的锁
                String token = stringRedisTemplate.opsForValue().get(SET_NX_EX_MENU_LIST_LOCK);
                if (uuid.equals(token)) {
                    // 确定是同一把锁， 才释放锁
                    stringRedisTemplate.delete(SET_NX_EX_MENU_LIST_LOCK);
                }
            }
            return update;
        } else{
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return updateMenuWithRedisLock(menu);
        }
    }
}

借助 lua 脚本我们成功将释放锁的操作变成原子性操作，确保了其正确性。

写到这里，我们对于分布式锁的来龙去脉，应该产生了一些属于自己的理解，现在就只剩下一个锁自动续期的问题没有解决了~

3.5、Redisson 实现分布式锁

锁需要续期，主要就是为了解决在一些业务场景中，业务执行超时，锁已经过期，但业务仍没执行完成的场景。

究其根本就是到底给锁多长时间算合适呢？这点其实是没法准确评估的。

如果不打算给锁自动续期的话，那么我觉得应当对锁的过期时间，适当的延长一些，以确保业务正确执行。

如果你和我一样是一名Java开发者，想要去实现锁自动续期，这个方案在市面上已经有成熟的轮子Redisson 啦~

在Redisson中，有一个著名的看门狗机制，当我们使用 Redisson 来实现分布式锁时，加锁时，每次都会默认设置过期时间30s，然后当业务执行超过10s，也就是锁时间还剩下 20s 时，它就会自动续期。

光说不练假把式，我们直接来用代码实现一下看看吧

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.17.6</version>
</dependency>

首先就是SpringBoot的起手式，编写MyRedissionConfig 类

@Configuration
public class MyRedissonConfig {

    /**
     * 所有对Redisson的使用都是通过RedissonClient
     * @return
     * @throws IOException
     */
    @Bean(destroyMethod="shutdown")
    public RedissonClient redissonClient() throws IOException {
        //1、创建配置
        Config config = new Config();
        // 连接必须要以 redis 开头~ 有密码填密码
        config.useSingleServer().setAddress("redis://IP地址:6379").setPassword("000415");
        //2、根据Config创建出RedissonClient实例
        //Redis url should start with redis:// or rediss://
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

我们所有的操作都是基于RedissonClient 来实现的，将它注入到Spring 容器中之后，在需要的时候直接引入即可。

另外 Redisson 它实现了 JUC 包下的大部分锁相关的实现，如果熟悉 JUC 的开发，使用这方面算是没什么学习成本的。如JUC 下的读写锁、信号量等。

我们就来使用一下Redisson中的读写锁吧~

/**
 * @description:
 * @author: Ning Zaichun
 * @date: 2022年09月20日 20:59
 */
@Service
public class RedissonServiceImpl implements IRedissonService {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MenuMapper menuMapper;

    private static final String REDISSON_MENU_LIST = "redisson:menu:list";
    private static final String REDISSON_MENU_LIST_LOCK_VALUE = "redisson:lock";
    @Autowired
    private RedissonClient redissonClient;

    @Override
    public List<MenuEntity> getList() {
        // 判断缓存是否有数据
        String menuJson = stringRedisTemplate.opsForValue().get(REDISSON_MENU_LIST);
        if (menuJson != null) {

            System.out.println("缓存中有，直接返回缓存中的数据");
            List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
            });
            return menuEntityList;
        }

        // 从数据库中查询
        List<MenuEntity> result = getMenuJsonFromDbWithRedissonLock();

        return result;
    }

    /**
     * 问题：其实写成上面那种模样，相对来说，也能解决很多时候的问题了，从头到尾看过来的话，其实也能发现就一个锁自动过期问题没有解决了。
     * 但在实现这个之前，我还是说明一下，为什么说在没有解决锁自动过期问题时，就已经能应付大多数场景了。
     * 重点在于如何评估 锁自动过期时间，锁自动过期时间到底设置多少合适呢？
     * 其实如果对于业务理解较为透彻，对于这一部分的业务代码执行时间能有一个较清晰的估算，给定一个合适的时间，在不出现极端情况，基本都能应付过来了。
     * <p>
     * 但是呢，很多时候，还是会怕这个万一的，万一真出现了，可能造成的损失就不止 一万了，哈哈。
     * 解决方案
     * 1、在市场主流的 Redission 中，针对这样的问题，已经有了解决方案。这也是Redission中常说的看门狗机制。
     * <p>
     * 如果需要自己实现的思路：
     * 1、这方面的问题，也做了十分浅显的思考，我觉得应该还是依赖于定时任务去实现，但到底该如何实现这个定时任务，我还没法给出一个合适的解决方案。 或许我应该会尝试一下。
     *
     * @return
     */
    public List<MenuEntity> getMenuJsonFromDbWithRedissonLock() {
        System.out.println("从数据库中查询");
        //1、占分布式锁。去redis占坑
        //（锁的粒度，越细越快:具体缓存的是某个数据，11号商品） product-11-lock
        //RLock catalogJsonLock = redissonClient.getLock("catalogJson-lock");
        //创建读锁
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(REDISSON_MENU_LIST_LOCK_VALUE);
        RLock rLock = readWriteLock.readLock();
        List<MenuEntity> result = null;
        try {
            rLock.lock();
            String menuJson = stringRedisTemplate.opsForValue().get(REDISSON_MENU_LIST);
            if (menuJson != null) {
                System.out.println("缓存中有，直接返回缓存中的数据");
                List<MenuEntity> menuEntityList = JSON.parseObject(menuJson, new TypeReference<List<MenuEntity>>() {
                });
                return menuEntityList;
            }
            try {
                Thread.sleep(50000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            //加锁成功...执行业务
            //加锁成功...执行业务
            result = menuMapper.selectList(new QueryWrapper<MenuEntity>());
            // 构建缓存
            stringRedisTemplate.opsForValue().set(REDISSON_MENU_LIST, JSON.toJSONString(result));
        } finally {
            rLock.unlock();
        }
        return result;
    }

    @Override
    public Boolean updateMenuById(MenuEntity menu) {
        return updateMenuWithRedissonLock(menu);
    }

    public Boolean updateMenuWithRedissonLock(MenuEntity menu) {
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(REDISSON_MENU_LIST_LOCK_VALUE);
        RLock writeLock = readWriteLock.writeLock();
        Boolean update = false;
        try {
            writeLock.lock();
            //加锁成功...执行业务
            //加锁成功...执行业务
            update = menuMapper.updateById(menu) > 0;
        } finally {
            writeLock.unlock();
        }
        return update;
    }

}

为了给大家测试一下它的自动续期，我在它第一次查询数据库获取锁时，让线程睡了一会，来进行测试。

测试：

[图片上传失败...(image-1e9d95-1666321506594)]

Redission的一些其他用法以及内部是如何实现锁续期的等等，这些都留在下一篇文章中啦。

3.6、关于Redission的一些补充

Redisson 默认锁过期时间 30s，一旦进行修改了的话，Redisson会取消此锁的自动续期机制。这一步的源码在 RedissonLock的tryAcquireAsync中

另外 Redisson 自动续期机制，也被大家称为看门狗机制，每次在锁还剩下20秒的时候，又会自动续到30s。默认时间在：

关于Redisson底层的一些流程分析，在明天的关于 Redisson 源码浅析的文章当中。

3.7、正确使用Redis的分布式事务锁

其实就下面两点：

上锁和设置过期时间需要保证原子性；（加锁）
判断锁ID是否为自己所有和解锁需要保证原子性（解锁）

保证这两步才能保证正确使用分布式锁，另外则是对于锁的过期时间需要进行一个合理评估，适当延长锁过期时间，如果需要实现锁自动续期可采用现有的轮子 Redisson 来实现。

四、小结

4.1、回顾

我们从本地锁一直讲到分布式锁，将Redis实现分布式锁中的一些问题，逐步进行了讲述。

从使用简单的 Redis 中的 SET KEY NX命令实现分布式锁
到使用SET KEY NX EX TIME 命令解决死锁问题
到增加身份标识（UUID）解决锁被其他人释放问题
再到使用 Lua 脚本，将解锁操作变成原子性操作
最后讲述了Redisson实现分布式锁，解决了锁自动续期问题

说它非常难的话，其实也没有，但是这是终点吗？

4.2、扩展

其实并不是，不知道你们有没有发现我上面所讲述的分布式锁，从始至终都是将Redis当作了一个实例来看待。

但在真正的环境中，Redis 远不止一个实例，部署方式也是多样的，主从复制、哨兵模式、集群模式，主从集群等等，那么在这些情况下，按照上面的方式去编写分布式锁会不会有问题呢？

你觉得呢？

答案是只要牵扯到网络通信，那么必然就会产生问题。（可以说在分布式中，网络永远都是处于个不可信的状态）

举个最简单的例子：

假设现在的部署方式是主从集群+哨兵模式，这样的好处是，当主库异常宕机时，哨兵可以实现「故障自动切换」，把从库提升为主库，继续提供服务，以此保证可用性。

那假设现在我一个请求进来，刚获取到锁，然后主节点就挂了，此时锁还没有同步到从节点上去，即使之后完成了主从切换，但是此次所加的锁也已经丢失。

因此在这样的基础上，Redis 官方继而又推出了 Redlock(红锁算法)。

五、关于红锁 Redlock

关于这些问题，Redis 的作者也提供了一些解决方案【Redlock】也就是我们常说的红锁。

红锁分析

Martin Kleppmann（英国剑桥大学的一名分布式系统研究员）关于 Redlock的分析
Redis 作者 Antirez 对于Martin Kleppmann的分析回复

两人的辩论都十分精彩，非常值得拜读，从中可以领略到诸多关于分布式的思考。

Redlock 简单使用来自 Redis 官网

5.1、红锁算法

在算法的分布式版本中，我们假设我们有 N 个 Redis master。这些节点是完全独立的，所以我们不使用复制或任何其他隐式协调系统。我们已经描述了如何在单个实例中安全地获取和释放锁。我们理所当然地认为算法会使用这种方法在单个实例中获取和释放锁。在我们的示例中，我们设置了 N=5，这是一个合理的值，因此我们需要在不同的计算机或虚拟机上运行 5 个 Redis 主服务器，以确保它们以几乎独立的方式发生故障。

为了获取锁，客户端执行以下操作：

它以毫秒为单位获取当前时间。
它尝试顺序获取所有 N 个实例中的锁，在所有实例中使用相同的键名和随机值。在步骤 2 中，当在每个实例中设置锁时，客户端使用一个与锁自动释放总时间相比较小的超时来获取它。例如，如果自动释放时间为 10 秒，则超时可能在 ~ 5-50 毫秒范围内。这可以防止客户端在尝试与已关闭的 Redis 节点通信时长时间保持阻塞：如果一个实例不可用，我们应该尽快尝试与下一个实例通信。
客户端通过从当前时间中减去步骤 1 中获得的时间戳来计算获取锁所用的时间。当且仅当客户端能够在大多数实例（至少 3 个）中获取锁时，且获取锁的总时间小于锁的有效时间，则认为锁已被获取。
如果获得了锁，则其有效时间被认为是初始有效时间减去经过的时间，如步骤 3 中计算的那样。
如果客户端由于某种原因未能获得锁（它无法锁定 N/2+1 个实例或有效时间为负数），它将尝试解锁所有实例（即使是它认为没有的实例）能够锁定）。

5.2、算法是异步的吗？

该算法依赖于这样一个假设，即虽然进程之间没有同步时钟，但每个进程中的本地时间以大致相同的速率更新，与锁的自动释放时间相比误差很小。这个假设非常类似于现实世界的计算机：每台计算机都有一个本地时钟，我们通常可以依靠不同的计算机来获得很小的时钟漂移。

在这一点上，我们需要更好地指定我们的互斥规则：只有持有锁的客户端在锁有效时间内（如步骤 3 中获得）内终止其工作，减去一些时间（仅几毫秒为了补偿进程之间的时钟漂移）。

本文包含有关需要绑定时钟漂移的类似系统的更多信息：租赁：分布式文件缓存一致性的有效容错机制。

5.3、失败重试

当客户端无法获取锁时，它应该在随机延迟后重试，以尝试使多个客户端同时尝试获取同一资源的锁（这可能导致没有人的脑裂情况）胜）。此外，客户端在大多数 Redis 实例中尝试获取锁的速度越快，裂脑条件的窗口就越小（并且需要重试），因此理想情况下，客户端应该尝试将SET命令发送到 N 个实例同时使用多路复用。

值得强调的是，对于未能获得大部分锁的客户端来说，尽快释放（部分）获得的锁是多么重要，这样就无需等待密钥到期才能再次获得锁（但是，如果发生网络分区并且客户端不再能够与 Redis 实例通信，则会在等待密钥到期时造成可用性损失）。

5.4、释放锁

释放锁很简单，无论客户端是否相信它能够成功锁定给定实例，都可以执行。

就是同时给所有的 Redis 实例发消息说要释放这把锁。

我此处只是一个简单的思路，如果对Redlock展开说，这篇文章的字数可能还需要翻上一倍，而且我感觉如果是没有实践的去分析，文字会稍显稚嫩，同时也会因为无案例支撑，让其真实性也会大打折扣。

想仔细了解的，大家可以多找找，网上也有很多针对两位大佬的辩论分析的文章。

5.5、补充

其实，如果你的应用只需要高性能的分布式锁并且可以接受一定程度上的数据不一致性【像之前说的刚设置完锁，Redis中的主机就宕机，导致没有成功同步到从机，所产生的数据不一致性】，那么实际上之前所讨论的 Redis 分布式锁也是足够了的。

但是如果业务要求一定要保证应用中数据的强一致性，那么我觉得你可以试着找找其他的方式，换成zookeeper 加上一定的补偿机制去试一试。毕竟Redlock(红锁)一方面太重了，不是特别大的项目，我个人觉得也不会用至少五台起步的Redis实例吧，另外一方面，看了两位大佬的讨论，特别极端的情况下，也是有可能出现问题的。

刚刚说到的Zookeeper 实现分布式锁,虽然它也有问题，但总归它是保证CAP机制中的CP的，可以保证任何时刻对Zookeeper的访问请求能得到一致性的数据，但不绝对保证服务一定可用~ 属于是用性能换安全啦~

总之，如果项目中一定要有非常强的数据一致性，在那么对于分布式锁，你也保留一丝怀疑，毕竟它也不是真的100%安全的。

既然看到这里啦，我觉得再看一遍大纲，判断一下自己理解了多少是非常重要的：

关于代码

不知道阅读的小伙伴，有木有发现代码中有一点点小问题~

后期在检查时，案例中的代码是有点不太合适的，应当将所有案例中的递归调用方法改为循环重试，并限制重试次数，而非一直递归调用。

原因：如果一直没有抢到锁，重复的递归调用是有很大可能会导致程序崩溃的,这是不合适的。另外如果是阅读过一些框架源码的话，它们的底层调用大都是写个while(true)来达到某个方法的重复调用，而并非是递归调用。

此处是我的疏忽，各位见谅见谅。

另外重试机制下可能会出现的问题：

幂等性问题：（查询操作具有天然幂等性）

在分布式架构下，服务之间调用会因为网络原因出现超时失败情况，而重试机制会重复多次调用服务，但是对于被调用放，就可能收到了多次调用。如果被调用方不具有天生的幂等性，那就需要增加服务调用的判重模块，并对每次调用都添加一个唯一的id。

大量请求超时堆积：

超高并发下，大量的请求如果都进行超时重试的话，如果你的重试时间设置不安全的话，会导致大量的请求占用服务器线程进行重试，这时候服务器线程负载就会暴增，导致服务器宕机。对于这种超高并发下的重试设计，我们不能让重试放在业务线程，而是统一由异步任务来执行。

作者：宁在春
链接：https://juejin.cn/post/7156623634327404557
来源：稀土掘金