聊到分布式锁，就不得不先聊到本地锁，如果没有从本地锁到分布式锁这个演进过程或者说是推导过程，我觉得是不合适的，甚至是不完整的。
程序的发展是一步一步递进，知道它是解决什么样的问题，才能更好的理解和学习。
上一篇文章：聊一聊缓存和数据库不一致性问题的产生及主流解决方案以及扩展的思考
本文就是针对上一篇讨论的数据一致性中的加锁方案来进行一个从浅至深的探究。
文章大纲：
本文字数约为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 也让锁得到共享，但是真的没问题吗？
思考🧐：
如果某个时刻代码执行到释放锁的那一步，服务突然挂了，是不是也就意味着锁释放失败了勒，那么也就造成了最大的问题死锁，因为没有解锁操作，锁也就将无法被释放。
那该如何解决呢？
最佳的方式就是给锁设置一个过期时间~
用万字长文来讲讲本地锁至分布式锁的演进和Redis实现，扩展 Redlock 红锁2：https://developer.aliyun.com/article/1394707
