分布式锁主流实现方案有哪些？（Redis分布式锁详细介绍原理和实现）

Redis 分布式锁

问题描述

1、单体单机部署的系统被演化成分布式集群系统后

2、由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效

3、单纯的Java API 并不能提供分布式锁的能力

4、为了解决这个问题就需要一种跨JVM 的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题

5、示意图(说明: 我们探讨的分布式锁是针对分布式项目/架构而言[.])

分布式锁主流实现方案

1. 基于数据库实现分布式锁
2. 基于缓存（Redis 等）
3. 基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

1. 性能：redis 最高
2. 可靠性：zookeeper 最高
3. 我们讲解基于Redis 实现分布式锁

实例： Redis 实现分布式锁-基本实现

指令: setnx key value

解读:

• setnx 可以理解是上锁/加锁指令
• key 是锁的键
• value 是锁的值
• 在这个key 没有删除前, 不能执行相同key 的上锁指令.

指令: del key

解读

• 就是删除key, 可以理解成就是释放锁

指令: expire key seconds

解读

• 给锁-key, 设置过期时间
• 目的是防止死锁

指令: ttl key

解读

• 查看某个锁-key, 过期时间

指令: set key value nx ex seconds

解读

• 设置锁的同时, 指定该锁的过期时间，防止死锁
• 这个指令是原子性的，防止setnx key value / expire key seconds 两条指令, 中间执行被打断.
• 过期时间到后, 会自动删除

实例： Redis 实现分布式锁-代码实现

需求说明/图解, 编写代码, 实现如下功能

1. 在SpringBoot+Redis 实现分布式锁的使用
2. 获取锁, key 为lock, 示意图

第1 种情况

–如果获取到该分布式锁

–就获取key 为num 的值, 并对num+1, 再更新num 的值, 并释放锁(key 为lock)

–如果获取不到key 为num 的值, 就直接返回

第2 种情况

–如果没有获取到该分布式锁

–休眠100 毫秒, 再尝试获取

在前面的SpringBoot 整合Reids 项目上实现即可

先在Redis 初始化数据

修改RedisTestController

\src\main\java\com\redis\controller\RedisTestController.java , 增加API 接口

java复制代码@GetMapping("testLock")
public void testLock() {
//1 获取锁，setnx
Boolean lock =
redisTemplate.opsForValue().setIfAbsent("lock", "ok");
//2 获取锁成功、查询num 的值
if (lock) {
Object value = redisTemplate.opsForValue().get("num");
//2.1 判断num 为空return
if (value == null || !StringUtils.hasText(value.toString())) {
return;
}
//2.2 有值就转成成int
int num = Integer.parseInt(value.toString());
//2.3 把redis 的num 加1
redisTemplate.opsForValue().set("num", ++num);
//2.4 释放锁，del
redisTemplate.delete("lock");
} else {
//3 获取锁失败、每隔0.1 秒再获取
try {
Thread.sleep(100);
testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

启动SpringBoot 项目

保证Linux 可以访问到SpringBoot 项目

使用ab 工具完成测试

ab -n 1000 -c 100 http://192.168.198.1:8080/redisTest/testLock

实例： 优化-设置锁的过期时间, 防止死锁

问题分析

假如在执行关闭锁之前就发生异常然后没有执行释放锁 然后我们又没有设置过期时间 所以就会死锁

1. 在前面代码上修改，设置锁的过期时间
2. 防止死锁

修改这一句就好了

注意这在实际开发中锁过期的时间是需要经过严格的考虑的不然设小了没有起效果 设置大了效率低

java复制代码Boolean lock =
redisTemplate.opsForValue().setIfAbsent("lock", "ok", 3, TimeUnit.SECONDS);

完成测试

ab -n 1000 -c 100 http://192.168.198.1:8080/redisTest/testLock

注意因为我们前面测试过一次了所以再测试就是2000次了

实例： 优化-UUID 防误删锁

问题分析, 如图

思路分析

1. 在获取锁的时候, 给锁设置的值是唯一的uuid
2. 在释放锁时,判断释放的锁是不是同一把锁.
3. 造成这个问题的本质原因, 是因为删除操作缺乏原子性

修改RedisTestController

java复制代码 @GetMapping("testLock")
public void testLock() {
//1 获取锁，setnx
//得到一个uuid 值，作为锁的值
String uuid = UUID.randomUUID().toString();
Boolean lock =
redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
//2 获取锁成功、查询num 的值
if (lock) {
Object value = redisTemplate.opsForValue().get("num");
//2.1 判断num 为空return
if (StringUtils.isEmpty(value)) {
return;
}
//2.2 有值就转成成int
int num = Integer.parseInt(value + "");
//2.3 把redis 的num 加1
redisTemplate.opsForValue().set("num", ++num);
//2.4 释放锁，del
//为了防止误删锁, 进行判断
//判断当前这个锁是不是前面获取到的锁, 相同才进行删除/释放
if (uuid.equals((String) redisTemplate.opsForValue().get("lock"))) {
redisTemplate.delete("lock");
}
//redisTemplate.delete("lock");
} else {
//3 获取锁失败、每隔0.1 秒再获取
try {
Thread.sleep(100);
testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

完成测试

ab -n 1000 -c 100 http://192.168.198.1:8080/redisTest/testLock

实例： 优化-LUA 脚本保证删除原子性

当前代码问题分析, 如图

思路分析

1. 删除操作缺乏原子性
2. 使用Lua 脚本保证删除原子性

修改RedisTestController

java复制代码@RestController
@RequestMapping("/redisTest")
public class RedisTestController {
//装配RedisTemplate
@Resource
private RedisTemplate redisTemplate;
//编写方法,使用Redis分布式锁,完成对 key为num的+1操作
@GetMapping("/lock")
public void lock() {
//得到一个uuid值,作为锁的值
String uuid = UUID.randomUUID().toString();
//1. 获取锁/设置锁 key->lock : setnx
Boolean lock =
redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
if (lock) {//true, 说明获取锁/设置锁成功
//这个key为num的数据，事先要在Redis初始化
Object value = redisTemplate.opsForValue().get("num");
//1.判断返回的value是否有值
if (value == null || !StringUtils.hasText(value.toString())) {
return;
}
//2.有值，就将其转成int
int num = Integer.parseInt(value.toString());
//3.将num+1,再重新设置回去
redisTemplate.opsForValue().set("num", ++num);
//释放锁-lock
//为了防止误删除其它用户的锁,先判断当前的锁是不是前面获取到的锁,如果相同，再释放
//=====使用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所对应的值
// 解读 Arrays.asList("lock") 会传递给 script 的 KEYS[1] , uuid 会传递给ARGV[1] , 其它的应该很容易理解
redisTemplate.execute(redisScript, Arrays.asList("lock"), uuid);
//if (uuid.equals((String) redisTemplate.opsForValue().get("lock"))) {
//    //...
//    redisTemplate.delete("lock");
//}
//redisTemplate.delete("lock");
} else { //获取锁失败,休眠100毫秒，再重新获取锁/设置锁
try {
Thread.sleep(100);
lock();//重新执行
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

Lua 脚本详解

完成测试

ab -n 1000 -c 100 http://192.168.198.1:8080/redisTest/testLock

注意事项和细节

1、定义锁的key, key 可以根据业务, 分别设置，比如操作某商品, key 应该是为每个sku 定义的，也就是每个sku 有一把锁

2、为了确保分布式锁可用，要确保锁的实现同时满足以下四个条件：

• 互斥性。在任意时刻，只有一个客户端能持有锁。
• 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
• 加锁和解锁必须是同一个客户端，A 客户端不能把B 客户端加的锁给解了
• 加锁和解锁必须具有原子性