本文转载自微信公众号“程序同事”，作者是楼下小黑哥。转载请联系项目总监公众号。重做总是比重建容易。最近在做一个项目，要把另一家公司的实现系统（以下简称老系统）完全集成到自己公司的系统（以下简称新系统）中，需要对方实现完整的功能也是在你自己的系统中实现。旧系统仍然有一些现有的商户。为了不影响现有商户的体验，新系统提供的对外接口必须与之前的保持一致。最后，系统完全切换后，功能只在新系统中运行，这就需要将旧系统中的数据完全迁移到新系统中。当然，这些都是在做这个项目之前就预料到的。本以为这个过程会很艰难，没想到这么艰难。本来觉得档期半年多，时间还是挺充裕的，现在感觉是个大坑，得一点一点的把坑填上。哎，说起来都是泪，我就不抱怨了，下次做完再给大家回顾一下我的真实经历。回到正文，在上一篇Redis分布式锁中，我们实现了基于Redis的分布式锁。这个分布式锁的基本功能没有问题，但是缺少可重入的特性，所以本文小黑就带大家实现一个可重入的分布式锁。本文将涵盖以下内容：ReentrantbasedonThreadLocalImplementationbasedonRedisHashImplementationreentrant说到可重入锁，首先我们来看看wiki上的一篇关于可重入的解释：“如果一个程序或者一个子程序可以”随时中断然后操作系统调度另一段代码去执行，而这段代码调用子程序没有错误”，所以称为可重入（reentrant或re-entrant）。即在子程序运行期间，执行线程可以重新进入进入并执行它仍然得到设计预期的结果。与多线程并发执行的线程安全不同，重入强调在单线程上执行时重新进入同一个子程序仍然是安全的。当一个线程执行一个块时代码并成功获取到锁，继续执行时，遇到被锁住的代码，重入保证线程可以继续执行，不可重入是指就是它需要等待锁被释放，可以再次成功获取到锁，然后可以继续执行。用一段Java代码来解释可重入：publicsynchronizedvoida(){b();}publicsynchronizedvoidb(){//pass}假设X线程在a方法获取锁b方法后继续执行，如果不是可重入这时，线程必须等待锁被释放，重新竞争锁。锁明明是X线程拥有的，但是还是需要等待自己释放锁，然后再去抢锁。这似乎很奇怪。我放飞自我~重入可以解决这个尴尬的问题。当线程拥有了锁之后，以后遇到加锁方法的时候，直接把加锁数加1，然后执行方法逻辑。退出加锁方法后，加锁数减1，当加锁数为0时，才真正释放锁。可以看出，可重入锁最大的特点就是计数，就是统计锁的数量。因此，在分布式环境中需要实现可重入锁时，我们还需要对锁的数量进行统计。分布式可重入锁的实现方式有两种：基于ThreadLocal的实现和基于RedisHash的实现。首先，让我们看一下基于ThreadLocal的实现。基于ThreadLocal实现的实现方法Java中的ThreadLocal允许每个线程都有自己的实例副本，我们可以利用这个特性来统计线程重入的次数。下面我们定义一个ThreadLocal全局变量LOCKS，内存存储Map实例变量。privatestaticThreadLocal>LOCKS=ThreadLocal.withInitial(HashMap::new);每个线程都可以通过ThreadLocal获取自己的Map实例，Map中的key存储锁的名称，value存储锁的重入次数。加锁的代码如下：/***Reentrantlock**@paramlockName锁名，代表需要竞争临界资源*@paramrequest唯一标识，可以使用uuid，根据这个值判断是否可以重入*@paramleaseTime解锁时间*@paramunit解锁时间单位*@return*/publicBooleantryLock(StringlockName,Stringrequest,longleaseTime,TimeUnitunit){Mapcounts=LOCKS.get();if(counts.containsKey(lockName)){counts.put(lockName,counts.get(lockName)+1);returntrue;}else{if(redisLock.tryLock(lockName,request,leaseTime,unit)){counts.put(lockName,1);returntrue;}}returnfalse;}"ps：redisLock#tryLock是上篇文章实现的分布式锁，由于公众号外链不能直接跳转，关注“流程”回复分布式锁获取源码，加锁方式首先判断当前线程是否被加锁，如果拥有锁，直接y将锁的重入次数加1。如果您不拥有锁，请尝试将其锁定在Redis中。加锁成功后，重入次数加1。释放锁的代码如下：/***解锁需要判断不同的线程池**@paramlockName*@paramrequest*/publicvoidunlock(StringlockName,Stringrequest){Mapcounts=LOCKS.get();if(counts.getOrDefault(lockName,0)<=1){counts.remove(lockName);Booleanresult=redisLock.unlock(lockName,request);if(!result){thrownewIllegalMonitorStateException("attempttounlocklock,notlockedbylockName:+"+lockName+"withrequest:"+request);}}else{counts.put(lockName,counts.get(lockName)-1);}}释放锁时，先判断重入次数。如果大于1，则表示锁为线程所有。所以直接把锁重入次数减1，如果当前重入次数小于等于1，先把Map中锁对应的key去掉，再去Redis去释放锁。这里需要注意的是，当锁不属于线程，直接解锁时，重入次数也小于等于1，此时也不一定能直接解锁成功。》在使用ThreadLocal的时候，记得及时清理内部存储实例变量，防止内存泄露，context数据字符串使用等问题，下次再说说最近使用ThreadLocal写的bug。相关问题，使用ThreadLocal记录一下本地重入次数，虽然确实简单高效，但是还是存在一些问题过期时间问题上面的加锁代码可以看到，重入加锁的时候，只对本地计数加1。这可能会导致一种情况，由于到业务执行时间过长，redis已经过期释放锁，再次重新入锁时，因为本地还有数据，所以认为锁还在持有，不符合实际情况。如果要在本地增加过期时间，还需要考虑本地是否和Redis过期时间一致，代码会变得很复杂。不同线程/进程的重入从狭义上讲，重入应该只是对同一个线程的重入，但实际业务可能会要求不同的应用线程可以重入同一个锁。ThreadLocal方案只能满足同一个线程的重入，不能解决不同线程/进程之间的重入问题。不同线程/进程的重入问题需要使用如下方案RedisHash方案来解决。在基于RedisHash可重入锁实现的ThreadLocal解决方案中，我们使用一个Map来记录锁的可重入次数，Redis也提供了一种可以存储键值对的Hash（哈希表）数据结构。所以我们可以利用RedisHash中保存的锁的重入次数，再通过lua脚本进行逻辑判断。加锁的lua脚本如下：----1代表true----0代表falseif(redis.call('exists',KEYS[1])==0)thenredis.call('hincrby',KEYS[1],ARGV[2],1);redis.call('pexpire',KEYS[1],ARGV[1]);return1;end;if(redis.call('hexists',KEYS[1],ARGV[2])==1)然后redis.call('hincrby',KEYS[1],ARGV[2],1);redis.call('pexpire',KEYS[1],ARGV[1]);return1;end;return0;"如果KEYS:[lock],ARGV[1000,uuid]不熟悉lua语言的同学，不用怕，上面的逻辑比较简单。加锁的代码首先使用Redis的exists命令判断当前锁是否存在，如果锁不存在，直接使用hincrby创建key为lock的哈希表，并将哈希表中的uuidkey初始化为0，然后再次加1，最后设置过期时间如果当前锁存在，则使用hexists判断当前锁对应的hash表中是否存在uuidkey，如果存在则再次使用hincrby加1，最后再次设置过期时间名词最后如果以上两种逻辑不匹配，直接返回。加锁代码如下：//初始化代码StringlockLuaScript=IOUtils.toString(ResourceUtils.getURL("classpath:lock.lua").openStream(),Charsets.UTF_8);lockScript=newDefaultRedisScript<>(lockLuaScript,Boolean.class);/***可重入锁**@paramlockName锁名，代表需要竞争临界资源*@paramrequest唯一标识，可以通过uuid判断是否可以重入*@paramleaseTime锁释放时间*@paramunit锁释放时间单位*@return*/publicBooleantryLock(StringlockName,Stringrequest,longleaseTime,TimeUnitunit){lonternalLockLeaseTime=unit.toMillis(leaseTime);returnstringRedisTemplate.execute(lockScript,Lists.newArrayList(lockName),String.valueOf(internalLockLeaseTime),request);}》Spring-Boot2.2.7.RELEASE只需要了解Lua脚本加锁逻辑，Java代码实现相当简单，直接使用SpringBoot提供的StringRedisTemplate即可。解锁的Lua脚本如下：--判断hashsetre-entrantkey的值是否等于0--如果为0则说明re-entrantkey不存在if(redis.call('hexists',KEYS[1],ARGV[1])==0)thenreturnil;end;--计算当前重入次数localcounter=redis.call('hincrby',KEYS[1],ARGV[1],-1);--小于等于0表示可以解锁if(counter>0)thenreturn0;elseredis.call('del',KEYS[1]);return1;end;returnnil;首先使用hexists判断给定字段是否存在于RedisHash表中。如果锁对应的Hash表不存在，或者keyuuid在Hash表中不存在，直接返回nil。如果存在，则说明当前锁被它持有。先用hincrby将重入次数减1，计算后再判断重入次数。如果小于等于0，则使用del删除锁。解锁后的Java代码如下：//初始化代码：StringunlockLuaScript=IOUtils.toString(ResourceUtils.getURL("classpath:unlock.lua").openStream(),Charsets.UTF_8);unlockScript=newDefaultRedisScript<>(unlockLuaScript,Long.class);/***Unlock*如果重入key的个数大于1，则重入key的个数减1
*unlocklua脚本的返回含义：
*1：表示解锁成功
*0：表示未释放锁，重入次数减1
*nil：表示有其他线程尝试解锁
**如果使用DefaultRedisScript，由于Spring-data-rediseval类型转换，
*Redis返回Nilbulk时，默认会转为false，会影响解锁语义，所以使用如下：
*DefaultRedisScript**具体转换代码请参考：
*JedisScriptReturnConverter
**@paramlockName锁名*@paramrequest唯一标识，可以使用uuid*@throwsIllegalMonitorStateException解锁前解锁，请先加锁。如果被锁住，解锁会抛出这个错误*/publicvoidunlock(StringlockName,Stringrequest){Longresult=stringRedisTemplate.execute(unlockScript,Lists.newArrayList(lockName),request);//如果没有返回值，说明是其他线程tryUnlockingif(result==null){thrownewIllegalMonitorStateException("attempttounlocklock,notlockedbylockName:+"+lockName+"withrequest:"+request);}}解锁代码的执行方式与加锁类似，只是返回类型不同解锁执行结果为Long。这里之所以没有像加锁一样使用布尔值，是因为在解锁lua脚本中，三个返回值的含义如下：1表示解锁成功，0表示加锁时重入次数减1被释放，null表示其他线程尝试解锁。如果解锁失败，如果返回值是Boolean，Spring-data-redis在进行类型转换时会将null转为false，影响我们的逻辑判断，所以返回类型必须是Long。以下代码来自JedisScriptReturnConverter：相关问题spring-data-redis低版本问题如果Spring-Boot使用Jedis作为连接客户端，使用RedisCluster集群模式，需要使用spring-boot-starter-data-redis2.1版本.9或更高版本，否则在执行过程中会抛出：org.springframework.dao.InvalidDataAccessApiUsageException:EvalShaisnotsupportedinclusterenvironment。如果当前应用无法升级spring-data-redis也没关系，可以通过以下方法直接使用原生Jedis连接执行lua脚本。以加锁代码为例：.getNativeConnection(),lockScript,Lists.newArrayList(lockName),Lists.newArrayList(String.valueOf(internalLockLeaseTime),reentrantKey));returnconvert(innerResult);});returnresult;}privateObjecteval(ObjectnativeConnection,RedisScriptredisScript,finalList<字符串>keys,finalListargs){ObjectinnerResult=null;//集群模式和单点模式执行脚本的方法相同，但是没有共同的接口，所以只能分开执行//Clusterif(nativeConnectioninstanceofJedisCluster){innerResult=evalByCluster((JedisCluster)nativeConnection,redisScript,keys,args);}//单点elseif(nativeConnectioninstanceofJedis){innerResult=evalBySingle((Jedis)nativeConnection,redisScript,keys,args);}returninnerResult;}数据类型转换问题如果使用Jedis原生连接执行Lua脚本，那么可能又会遇到数据类型转换的坑。可以看到Jedis#eval返回的是Object，我们需要根据Lua脚本的返回值进行相关的转换。这涉及将Lua数据类型转换为Redis数据类型。下面主要说说Lua数据到Redis转换的一些比较容易踩的规则：1.Lua数和Redis数据类型转换。Lua中的数字类型是双精度浮点数，而Redis只支持整型，所以这个转换过程会去掉小数位。2.Luaboolean和Redis的类型转换这个转换比较容易踩。Redis中没有boolean类型，所以在Lua中为true时会转为Redis整数1。在Lua中为false不转为整数，而是转为null返回给客户端。3、Luanil与Redis类型转换Luanil可以看做是一个null值，可以等同于Java中的null。在Lua中，如果nil出现在条件表达式中，它将被视为false。所以Luanil也会返回null给客户端。其他转换规则比较简单，详见：http://doc.redisfans.com/script/eval.html，基于ThreadLocal的实现方案，实现简单，运行效率更高。但是要处理锁过期的问题，代码实现比较复杂。另一种使用RedisHash数据结构的实现方案解决了ThreadLocal的缺陷，但代码实现难度稍大，需要熟悉Lua脚本和一些Redis命令。另外，在使用spring-data-redis操作Redis时，不经意间会遇到各种问题。帮助https://www.sofastack.tech/blog/sofa-jraft-rheakv-distributedlock/https://tech.meituan.com/2016/09/29/distributed-system-mutually-exclusive-idempotence-cerberus-gtis.html
