推荐文章：
1、springBoot对接kafka,批量、并发、异步获取消息,并动态、批量插入库表;
2、SpringBoot用线程池ThreadPoolTaskExecutor异步处理百万级数据;
3、java后端接口API性能优化技巧
4、SpringBoot+MyBatis流式查询,处理大规模数据,提高系统的性能和响应能力;
5、SpringBoot整合多数据源，并支持动态新增与切换（详细教程）
一、为什么需要分布式锁?
传统单体/集群开发都是 Jvm 进程内的锁如：lock锁，synchronized锁，再比如cas原子类轻量级锁，但是对于跨 Jvm 进程以及跨机器，这种锁就不适合业务场景，会存在问题。并且JDK原生的锁可以让不同线程之间以互斥的方式来访问共享资源，但若想要在不同进程之间以互斥的方式来访问共享资源，JDK原生的锁就无能为力了(对于多线程程序，避免同时操作一个共享变量而产生数据问题，我们通常会使用一把锁来互斥以保证共享变量的正确性，其使用范围是在同一个进程中，如果换做是多个进程，需要同时操作一个共享资源，如何互斥呢？现在的业务应用通常是微服务架构，这也意味着一个应用会部署多个进程，例如：多个进程如果需要修改MySQL中的同一行记录，多个进程同时启动定时任务更新数据等等，为了避免操作乱序导致脏数据，此时就需要引入分布式锁了)。
因此，想要实现分布式锁，须借助一个外部系统，所有进程都去这个系统上申请加锁。而这个外部系统，必须要有互斥能力，即：两个请求同时进来的时候，只能给一个进程加锁成功，另一个失败。这个外部系统可以是数据库，也可以是Redis或Zookeeper，但考虑到性能，我们通常会选择使用Redis或Zookeeper来实现。
二、Redis分布式锁如何实现？
核心思想：set ex px nx + 校验唯一随机值,再删除
若实现分布式锁，必须要求Redis有互斥的能力。
Redis实现分布式锁的核心命令如下：
SETEX key valueSETEX：SET IF NOT EXIST，如果指定的key不存在，则创建并为其设置值，然后返回状态码1；如果指定的key存在，则直接返回0。如果返回值为1，代表获得该锁；此时其他进程再次尝试创建时，由于key已经存在，则都会返回0，代表锁已经被占用。
// 1、加锁SETNX lock_key 1// 2、实现业务逻辑DO THINGS// 3、释放锁DEL lock_key
当获得锁的进程处理完成业务后，再通过del命令将该key删除，其他进程就可以再次竞争性地进行创建，获得该锁。但是存在以下问题：
1、程序处理步骤二：实现业务逻辑发生异常，没及时释放锁;
2、进程挂了，没机会释放锁。
以上情况会导致已经获得锁的客户端一直占用锁，其他客户端永远无法获取到锁，也即“死锁”。
三、如何解决死锁问题？
最容易想到的方案是在申请锁时，在Redis中实现时，给锁设置一个过期时间，假设操作共享资源的时间不会超过10s，那么加锁时，给这个key设置10s过期即可。在Redis 2.6.12之后，Redis扩展了SET命令的参数，可以在SET的同时指定EXPIRE时间，这条操作是原子的，例如：以下命令是设置锁的过期时间为10秒。
命令：SET KEY VALUE [EX seconds] [PX milliseconds] [NX|XX]例如：SET lock_key 1 EX 10 NX
-
EX seconds-设置指定的终止时间，以秒为单位。
-
PX milliseconds-设置指定的终止时间（以毫秒为单位）。
-
NX - 仅在不存在的情况下设置key。
-
XX - 仅设置key（如果已存在）。
但是，还有锁过期/释放了别人的锁问题：
1、线程1加锁成功，开始操作共享资源;
2、线程1操作共享资源耗时太久，超过了锁的过期时间，锁失效（锁被自动释放）；
3、线程2加锁成功，开始操作共享资源；
4、线程1操作共享资源完成，在finally块中手动释放锁，但此时它释放的是客户端2的锁。
问题分析：
1、锁过期问题：评估操作共享资源的时间不准确导致的，若只是增大过期时间，只能缓解问题降低出现问题的概率，仍然无法彻底解决问题。原因在于客户端在拿到锁之后，在操作共享资源时，遇到的场景是很复杂的，既然是预估的时间，也只能是大致的计算，不可能覆盖所有导致耗时变长的场景。
2、释放了别人的锁问题：原因在于释放锁的操作并没有检查这把锁的归属，这样解锁不严谨。
四、如何避免锁被别人给释放？
客户端在加锁时，设置一个只有自己才知道的唯一标识进去，例如：可以是自己的线程ID/UUID产生的值，之后在释放锁时，要先判断这把锁是否归自己持有，只有是自己的才能释放它。
4.1、使用Lua脚本
//释放锁 比较unique_value是否相等，避免误释放if redis.get("key") == unique_value(线程ID/UUID产生的值) thenreturn redis.del("key")
GET + DEL两个命令需要使用Lua脚本，保证原子的执行。因为Redis处理每个请求是单线程执行的，在执行一个Lua脚本时其它请求必须等待，直到这个Lua脚本处理完成，这样一来GET+DEL之间就不会有其他命令执行了。
unlock.script脚本如下：
//Lua脚本语言，释放锁 比较unique_value是否相等，避免误释放if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])elsereturn 0end
其中，KEYS[1]：lock_key，ARGV[1]：当前客户端的唯一标识，这两个值都是我们在执行 Lua脚本时作为参数传入的。
在java代码中的运用：
/*** 解锁脚本，原子操作*/private static final String unlockScript ="if redis.call(\"get\",KEYS[1]) == ARGV[1]\n"+ "then\n"+ " return redis.call(\"del\",KEYS[1])\n"+ "else\n"+ " return 0\n"+ "end";
使用：
/**
* 功能描述:使用Lua脚本解锁
* @MethodName: unlock
* @MethodParam: [name, token]
* @Return: boolean
* @Author: yyalin
* @CreateDate: 2023/7/17 18:41
*/
public boolean unlock(String name, String token) {
byte[][] keysAndArgs = new byte[2][];
keysAndArgs[0] = name.getBytes(Charset.forName("UTF-8")); //lock_key
keysAndArgs[1] = token.getBytes(Charset.forName("UTF-8")); //token的值，也即唯一标识符
RedisConnectionFactory factory = redisTemplate.getConnectionFactory();
RedisConnection conn = factory.getConnection();
try {
Long result = (Long)conn.scriptingCommands().eval(unlockScript.getBytes(Charset.forName("UTF-8")),
ReturnType.INTEGER, 1, keysAndArgs);
if(result!=null && result>0)
return true;
}finally {
RedisConnectionUtils.releaseConnection(conn, factory);
}
return false;
}
五、代码实现
5.1、RedisLock类
/**
* 功能描述:redis的分布式锁：解决并发问题
* @Author: yyalin
* @CreateDate: 2023/7/18 10:17
*/
@Repository
public class RedisLock {
/**
* 解锁脚本，原子操作
*/
private static final String unlockScript =
"if redis.call(\"get\",KEYS[1]) == ARGV[1]\n"
+ "then\n"
+ " return redis.call(\"del\",KEYS[1])\n"
+ "else\n"
+ " return 0\n"
+ "end";
private StringRedisTemplate redisTemplate;
//有参构造函数
public RedisLock(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
}
/**
* 加锁，有阻塞
* @param name key的值
* @param expire 过期时间
* @param timeout 加锁执行超时时间
* @return
*/
public String getLock(String name, long expire, long timeout){
long startTime = System.currentTimeMillis(); //获取开始时间
String token;
//规定的时间内，循环获取有值的token
do{
token = tryGetLock(name, expire); //获取秘钥Key
if(token == null) {
if((System.currentTimeMillis()-startTime) > (timeout-50))
break;
try {
Thread.sleep(50); //try 50毫秒 per sec milliseconds
} catch (InterruptedException e) {
e.printStackTrace();
return null;
}
}
}while(token==null);
return token;
}
/**
* 加锁，无阻塞
* @param name 设置key
* @param expire
* @return
*/
public String tryGetLock(String name, long expire) {
//获取UUID值为value
String token = UUID.randomUUID().toString();
RedisConnectionFactory factory = redisTemplate.getConnectionFactory();
RedisConnection conn = factory.getConnection();
try{
Boolean result = conn.set(name.getBytes(Charset.forName("UTF-8")), //设置name为key
token.getBytes(Charset.forName("UTF-8")), //设置token为value
Expiration.from(expire, TimeUnit.MILLISECONDS), //设置过期时间：MILLISECONDS毫秒
RedisStringCommands.SetOption.SET_IF_ABSENT); //如果name不存在创建
if(result!=null && result)
return token;
}finally {
RedisConnectionUtils.releaseConnection(conn, factory);
}
return null;
}
/**
* 功能描述:使用Lua脚本解锁
* @MethodName: unlock
* @MethodParam: [name, token]
* @Return: boolean
* @Author: yyalin
* @CreateDate: 2023/7/17 18:41
*/
public boolean unlock(String name, String token) {
byte[][] keysAndArgs = new byte[2][];
keysAndArgs[0] = name.getBytes(Charset.forName("UTF-8")); //lock_key
keysAndArgs[1] = token.getBytes(Charset.forName("UTF-8")); //token的值，也即唯一标识符
RedisConnectionFactory factory = redisTemplate.getConnectionFactory();
RedisConnection conn = factory.getConnection();
try {
Long result = (Long)conn.scriptingCommands().eval(unlockScript.getBytes(Charset.forName("UTF-8")),
ReturnType.INTEGER, 1, keysAndArgs);
if(result!=null && result>0)
return true;
}finally {
RedisConnectionUtils.releaseConnection(conn, factory);
}
return false;
}
}
5.2、控制层调用
@ApiOperation(value="添加学生", notes="add")@PostMapping("/add")public void addStudent() throws InterruptedException {String token = null;try{//设置锁并获取唯一值token = redisLock.getLock("lock_name", 10*1000, 11*1000);if(token != null) {System.out.println("我拿到了锁哦:"+token);// 开始执行业务代码Thread.sleep(3*1000L);} else {System.out.println("我没有拿到锁唉");//1000毫秒后过一会在尝试重新获取锁Thread.sleep(5*1000L);System.out.println("我开始重试来了。。。。。");addStudent();}}finally {if(token!=null) {//用完进行释放锁redisLock.unlock("lock_name", token);}}}
六、总结
基于Redis实现的分布式锁，一个严谨的流程如下：(set ex px nx + 校验唯一随机值,再删除)
1、加锁时要设置过期时间SET lock_key unique_value EX expire_time NX
2、操作共享资源(业务代码);
3、释放锁：Lua脚本，先GET判断锁是否归属自己，再DEL释放锁。
到此原生的redis实现分布式加锁、解锁流程就更加严谨了，可以满足大部分场景,用来解决大部分的并发问题。
更多详细资料，请关注个人微信公众号或搜索“程序猿小杨”添加。
参考：
https://huaweicloud.csdn.net/63355ebdd3efff3090b546db.html?spm=1001.2101.3001.6650.1&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7Eactivity-1-119251590-blog-127391210.235%5Ev38%5Epc_relevant_sort_base1&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7Eactivity-1-119251590-blog-127391210.235%5Ev38%5Epc_relevant_sort_base1&utm_relevant_index=2
