一、分布式锁概念
随着业务发展的需要，原单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的 Java API 并不能提供分布式锁的能力。为了解决这个问题就需要一种跨 JVM 的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！
说得通俗些，集群中上了锁后，无论当前操作在哪台机器，所有的机器都会识别并且等待，锁释放后其他操作才能进行，这就是分布式锁，对所有集群里都有效
分布式锁主流的实现方案：
- 基于数据库实现分布式锁
- 基于缓存（Redis 等）
- 基于 Zookeeper
每一种分布式锁解决方案都有各自的优缺点，其中redis性能最高zookeeper可靠性最高
二、使用setnx实现锁
set stu:1:info “OK” nx px 10000
- EX second ：设置键的过期时间为 second 秒,，SET key value EX second 效果等同于 SETEX key second value
- PX millisecond ：设置键的过期时间为 millisecond 毫秒，SET key value PX millisecond 效果等同于 PSETEX key millisecond value
- NX ：只在键不存在时，才对键进行设置操作，SET key value NX 效果等同于 SETNX key value
- XX ：只在键已经存在时，才对键进行设置操作
- 多个客户端同时获取锁（setnx）
- 获取成功，执行业务逻辑（从 db 获取数据，放入缓存），执行完成释放锁（del）
- 获取失败的客户端则等待重试
用setnx和del添加以及释放锁
一般地，我们需要给锁设置过期时间防止锁被长期占用
这里有个问题：加锁和设置过期时间是两个操作，而不是同时进行操作的，如果上锁后发生异常情况，就无法设置过期时间了。我们可以上锁的同时设置过期时间
三、编写代码测试分布式锁
1. 使用Java代码测试分布式锁
首先在redis中设置num的值为0，编写Java代码进行测试
下方代码做的就是：获取到锁则num++，并释放锁；没获取到则0.1秒后重新获取
重启，服务集群，通过网关压力测试：ab -n 5000 -c 100 http://192.168.140.1:8080/test/testLock
查看 redis 中 num 的值
问题： setnx 刚好获取到锁，业务逻辑出现异常，导致锁无法释放
解决： 设置过期时间，自动释放锁
2. 优化之设置锁的过期时间
设置过期时间有两种方式：
- 首先想到通过 expire 设置过期时间（缺乏原子性：如果在 setnx 和 expire 之 间出现异常，锁也无法释放）
- 在 set 的同时指定过期时间（推荐）
代码中设置过期时间：
问题： 可能会释放其他服务器的锁
如果业务逻辑的执行时间是 7s，执行流程如下：
- index1 业务逻辑没执行完，3 秒后锁被自动释放
- index2 获取到锁，执行业务逻辑，3 秒后锁被自动释放
- index3 获取到锁，执行业务逻辑
- index1 业务逻辑执行完成，开始调用 del 释放锁，这时释放的是 index3 的锁， 导致 index3 的业务只执行 1s 就被别人释放。
最终等于没锁的情况
a在操作时卡顿了，导致锁超时后自动释放；释放后，b抢到锁进行操作；此时a操作完成，手动释放锁，这就把b的锁给释放了，b再释放锁则会报错
解决： setnx 获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这 个值，判断是否自己的锁
四、优化之给lock设置UUID防误删
五、使用LUA脚本保证删除的原子性
使用lock的uuid可以一定程度上缓解线程释放其他锁，但并不能完全解决这种情况。因为比较uuid和删除lock并不是原子性的
问题： a比较uuid通过后，锁到期了自动释放，b重新加锁，a此时会手动释放b的锁，这还是出现问题
解决： 使用LUA 脚本保证删除的原子性
LUA脚本：
- 将复杂的或者多步的 redis 操作，写为一个脚本，一次提交给 redis 执行，减少反复连接 redis 的次数，提升性能
- LUA 脚本是类似 redis 事务，有一定的原子性，不会被其他命令插队，可以完成一些redis 事务性的
@GetMapping("testLockLua") public void testLockLua() { //1 声明一个 uuid ,将做为一个 value 放入我们的 key 所对应的值中 String uuid = UUID.randomUUID().toString(); //2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！ String skuId = "25"; // 访问 skuId 为 25 号的商品 100008348542 String locKey = "lock:" + skuId; // 锁住的是每个商品的数据 // 3 获取锁 Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS); // 第一种： lock 与过期时间中间不写任何的代码。 // redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间 // 如果 true if (lock) { // 执行的业务逻辑开始 // 获取缓存中的 num 数据 Object value = redisTemplate.opsForValue().get("num"); // 如果是空直接返回 if (StringUtils.isEmpty(value)) { return; } // 不是空 如果说在这出现了异常！ 那么 delete 就删除失败！ 也就是说锁永远存在！ int num = Integer.parseInt(value + ""); // 使 num 每次+1 放入缓存 redisTemplate.opsForValue().set("num", String.valueOf(++num)); /*使用 lua 脚本来锁*/ // 定义 lua 脚本：将判断和删除操作同时进行 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; // 使用 redis 执行 lua 执行 DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(); redisScript.setScriptText(script); // 设置一下返回值类型 为 Long // 因为删除判断的时候，返回的 0,给其封装为数据类型。如果不封装那么默认返回 String 类型， // 那么返回字符串与 0 会有发生错误。 redisScript.setResultType(Long.class); // 第一个是执行的 script 脚本 ，第二个需要判断的 key，第三个就是 key 所对应的值。 redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid); } else { // 其他线程等待 try { // 睡眠 Thread.sleep(1000); // 睡醒了之后，调用方法。 testLockLua(); } catch (InterruptedException e) { e.printStackTrace(); } } }
为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：
- 互斥性；在任意时刻，只有一个客户端能持有锁
- 不会发生死锁；即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁（设置lock的过期时间）
- 解铃还须系铃人；加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了（使用LUA脚本和uuid）
- 加锁和解锁必须具有原子性（使用LUA脚本）
