随着业务发展的需要,原单机部署的系统被演化成分布式集群系统后,由于分布式系统多线程、多进程并且分布在不同机器上,这将使原单机部署情况下的并发控制锁策略失效,单纯的 Java API 并不能提供分布式锁的能力。为了解决这个问题就需要一种跨 JVM 的互斥机制来控制共享资源的访问,这就是分布式锁要解决的问题!
说得通俗些,集群中上了锁后,无论当前操作在哪台机器,所有的机器都会识别并且等待,锁释放后其他操作才能进行,这就是分布式锁,对所有集群里都有效
分布式锁主流的实现方案:
每一种分布式锁解决方案都有各自的优缺点,其中redis性能最高zookeeper可靠性最高
set stu:1:info “OK” nx px 10000
用setnx和del添加以及释放锁
一般地,我们需要给锁设置过期时间防止锁被长期占用
这里有个问题:加锁和设置过期时间是两个操作,而不是同时进行操作的,如果上锁后发生异常情况,就无法设置过期时间了。我们可以上锁的同时设置过期时间
首先在redis中设置num的值为0,编写Java代码进行测试
下方代码做的就是:获取到锁则num++,并释放锁;没获取到则0.1秒后重新获取
重启,服务集群,通过网关压力测试:ab -n 5000 -c 100 http://192.168.140.1:8080/test/testLock
查看 redis 中 num 的值
问题: setnx 刚好获取到锁,业务逻辑出现异常,导致锁无法释放 解决: 设置过期时间,自动释放锁
设置过期时间有两种方式:
代码中设置过期时间:
问题: 可能会释放其他服务器的锁
如果业务逻辑的执行时间是 7s,执行流程如下:
a在操作时卡顿了,导致锁超时后自动释放;释放后,b抢到锁进行操作;此时a操作完成,手动释放锁,这就把b的锁给释放了,b再释放锁则会报错
解决: setnx 获取锁时,设置一个指定的唯一值(例如:uuid);释放前获取这 个值,判断是否自己的锁
使用lock的uuid可以一定程度上缓解线程释放其他锁,但并不能完全解决这种情况。因为比较uuid和删除lock并不是原子性的
问题: a比较uuid通过后,锁到期了自动释放,b重新加锁,a此时会手动释放b的锁,这还是出现问题
解决: 使用LUA 脚本保证删除的原子性
LUA脚本:
@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();
}
}
}
为了确保分布式锁可用,我们至少要确保锁的实现同时满足以下四个条件: