面试官：如何用Redis实现分布式锁？

点赞再看，养成习惯，微信查找【三太子敖丙】关注这个互联网苟全性命的东西人。

本文 GitHub github.com/JavaFamily 已录入，有一线大厂面试完好考点 1 w 4 O、材料以及我的系列文章。

前语

上一章节我提到了基于zk分布式锁的完成，这章节就来说一下基于Redis的分布式锁完成吧i – s。

• zk完成分布式锁的传送门：zk分= U X g –布式锁I S S t z L 9 } h

在开端提到Redis分布式锁之前，我想跟咱们聊点Redis的基础知识。

说一下Redis的两个指令：

SETNX key value

set, I @ ynx 是SETd W 0 a ( : if Not eXists(假如不存在，则 SET)的简写。

用法如图，假如不存在set成功回来int的1，这I ? v } 个key4 f , B N 3 !存在了回来0。

SETEX key seconds value

将值 value 关联到 key ，并将 key 的生存时刻设为 seconds (以秒为单位)。

假如 key 现已存在，setex指令将覆写旧值。

有小伙伴必定会疑惑万一set valuW j # & ; L J Z ae 成功 set time失利，那不就傻了么，这啊R? ~ 7edis官网想到了。

setex是一个原子性(atomic)操作，关联值和设置生存时刻两个动作会在同一时刻内完成。

我设置了10秒的失效时刻，ttl指令能够查看倒计时，负的阐明现已到期了。

跟咱们讲这两个命名也是有原因的，由于他们是Redis完成分布式锁的要害。

正文

开端前仍是看看场景:

我依然是创建了许多个线程去扣减库存inventory，不出意外的库存扣减次序变了，最终的成果也是不对的。

单机加synchronized或者Lock这些惯例操作我就不说了好吧，成果必定是对的。

我先完成一个简略的Redis锁，然后咱们再完成分布式锁，V ^ U或许更便利咱们的理解。

还记得上面我说过的指令么，完成一个单机的其实比较简略，你们先思考一下，别往下看。

setnx

能够看到，第一个成功了，没开释锁，后边的都失利了，至少次序问题问题是解决了，只需加锁，缩放= @ 9 f后边的拿到，开释如此循环，就能确保依照次序履行。

可是你们也发现问题了，仍是相同的，第一个仔set成功了，可是忽然挂了，那锁就一向在那无法得到开释，后边的线程也永远得不到锁，又死锁了。

所以….

setex

知道我之前说这个指令的[ 4 y s原因了吧，设置一个过期时刻，就算线程1挂了，也会在失效时刻到了，自动开释。

我这儿就用到了nx和px的结合参数，便是set值并且加了过期时刻，这儿我还t # ~ { Q P设置了一个过期时刻，便是这时刻内假如第二个没拿到第一个的锁，就退出堵塞了，由于或许是客户端断连了。

加锁r H H ~ p O s

整体T ) ^ o Z Y R加锁的逻辑比较简略，咱们基本上都能看懂，不过我拿到当时时刻去减开端时刻的操作感觉有点笨，[ 8 X G System.cuE j s T X W ] K ErrentTimeMillis()消耗很大的。

/d a X }**
*加锁
*
*@paramid
*@retu@ ? c 3 ? b c { Krn
*/
publicbooleanlock(Stringid){
Longstart=System.curreG  h j Z QntTimeMillis();
try{
for(;;){
//SET指令回来OK，则证明获取锁成功
StringlocF p ]k=jedis.set(LOCK_KEY,id,params);
if("OK".equals(loZ q q u K :ck} C S 3 x W x d)){
returntruea ; [ { q;
}
//不然循环等候，在timeout时刻内仍未获取到锁，则获取失利
lo$ K O 8 ]ngl=System.currentTimeMilV ( $ 0 k 3 P glis()-start;
if(l>=timeout){
returnfalse;
}
try{
Thread.slA S * ] q 8eep(100);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}finally{
jeda . t Iis.close();
}
}

System.currentTimeMillis消耗大，每个线程进来都这样，我之前写F ; &代码，就会在服务器发动的时分，开一个H , { v j 7 (线程不断去拿，调用方直接获取值就好了，不过也不是最优解，日期I E . X u X ) (类仍是有许多好办法的。

@Service
publicclassTiy i ; - # 2meServcie{
priva) C M Z O ctestaticlongtime;
static{
newThread(newRunnable(){
@Override
publicvoidrun(){
while(true){
try{
Thread.sleep(5);
}catch(InterruptedExcep~ $ s h w * xtione){
e.printStackTrace();
}
longcur=System.currentTimeMillis();
setTime(cur);
}
}
}).start();
}

publicstaticlonggetTime(){
returnth ] i F f +ime;
}

publicstaticvoidsetTime(longtime){
TimeServcie.time=time;
}
}

解锁

解锁的逻辑愈加简略，便是一段L= L ) [ 4 )ua的拼装，把Key做了删去。@ n !

你们发现没，我上面加锁解锁都0 ; 8 2 + . X (用了UUID，这便是为c 6 p B ] z J 9了确保，谁加锁了谁解锁，要是你删掉了我的锁，那不乱套了嘛。

LUA是原子性的，也比较简略，: & u 5 #便是判别一下Key和咱们参数是否相等，, % ]是的话就删去，回来成功1，0便是失利。

/**
*解锁
*
*@parb K @ O u namid
*@return
*/
publicbooleanunlock(Str~ 0 iingid){
Stringscript=
"ifredis.call('get',KEY7 3 C # X C Y 9S[1])==ARGV[1]then"G ^ - 4 J 7 f x B+
"returnredis.call('del',KEYS[1])"+
"else"+
[ w 9 s"return0"+
"endF ] i ! E k";
try{
Stringresult=jedis.eval(script,Collections.singletonList(LOCK_KEY),Collections.singletonList(id)).toString();
return"1".equals(result)?true:false;
}finally{
jedis.close();
}
}

验证

咱们能够用咱们写的Redis锁试试作用，能够看到都依照次序去履行了

思考

咱们是不是觉得完美了，可是上面的锁，有不少瑕疵的，我没思考许多点，你或许能够思考一下，源码我都开# % l源到我的GItHub了。

并且，锁一般都是需求可重入行的，上面的线程都是履行完了就开释了，无法再次进入了，进去J n , Z i [ V也是从头加锁了，对于一个锁的规划来说必定不是很合理的。

我不打算手写，由于都有现成的，别人帮咱们写好了。

redisson

rex 9 h 2 M Y Tdisson$ / 2 G [的锁，就完成了可重入了，可是他的源码比较不流畅难明。

使用起来很简略，由于他们底层都封装好了，你连接上你的Redis客户端，他帮你做了我上面写的一切，然后更完美。

简略看看他的使用吧，跟正常使用Lock没啥差异。

ThreadPoolExecut: = r F p l t =orthreadPC q n  goolExecutor=
newT4 F 7 J threadPoolExecutor(inventory,inventory,10L,SECONDS,linkedBlockingQueue);
l| j K , F A y songstart=System.currentTimeMillis();
Configconfig=newConfig();
config.useSingleServer().setAddress("reD V 6 C w Bdis://127.0.0.1:6379")a % o # & | T;
finalRedissonCliY 6 Qentclient=Redisso{ ] R I w T v Fn.create(0 v E r . Rconfig);
finalRLocklocS S % (k=client.getLock("lock1");

for(inti=0;i<=NUM;i+d T 6 = z+){
threadPool& b U j G  rExecutor.execute[ ~ F v e d(newRunnable(){
publicvoidrun(){
lock.lock();
inventory--;
System.out.println(inventory);
lock.unlock();
}
});
}
long! n M 9 ) mend=System.currentTimeMillis();
System.out.println("履行线程数:"+NUM+"总耗时:"+(end-start)+"库存数为:"+inventory);

上面能够看到我用到了getLockH 6 c W H [ C，其实便E { I是获取一个锁的实例。

RedissionLock也没做啥，便是了解的初始化。

publicRLockgetLock(Stri& } # X *ngname){
returnnewRedissonLock(connectionManager.getCommandExecutor(),name);
}

publicRedissonLock(CommandAsyncExecutorcomma3 + C 3 5 V [ a mndExecutor,Stringname){
super(commandExecutor,name);
//指令履行器
this.commandExecutor=commandExo / Becutor;
//UUID字符串
this.id=commandExecutor.getConnectionManager().getI7 i [d();
//内部锁过期时刻
this.internalLockLeaseTime=commandEE h f dxecutor.
getConnectir c { 4 I m 8 l DonManager().getCfg().getLockWatchdogTimeout( . # o E x ? C Z);
this.entryName=id+":"+name;
}

加锁

有没有发现许多跟Lock许多相似的地方呢？

测验加锁，拿到当时线程，然后我开头说的tV J C + J t $tl也看到了，是不是一切都是那么了解？

publicvoidlockInterruptibly(longleaseTime,TimeUnit l Uunit)throwsInte& 0 X g 8 M j Q rruptedException{

//当时线程ID
longthreadId=Thread.curry 2 B [entThread().getId();
//J W T ( 0 A %测验获取锁= t S d m
Longttl=tryAcquire(leaseTime,unit,threadId);
//假如ttl为空，则证明获取4 % c k 1 ^ 锁成功
if(^ f U r 2 g httl==null){
retu@ X l g : k Srn;
}
//假如获取锁失利，则订阅到对应这个锁的channel
RFuture<RedissonLockEntry&g& M ` Kt;fum U y c l & Q :tu8 $ . are=subscribe(threadId);
commandExecg z V W = z wutor.syncSubscrL L J r K 7 ription(future);

try{
while(true){
//再6 7 3 k V N z次测验获取锁
ttl=tryAcquire@ | * M W s w |(leaseTime,unit,threadId);
//ttl为空，阐明成功获取锁，回来6 8 p 8 D 9
if(ttl==null){
break;
}
//ttl大于0则等候ttl时刻后继续测验获取
if(ttl>=0){
getEntry0 Z h J W P(threadI` 2 A nd).getLatch()./ % atryAcquire(ttl,T/  O , * F p DimeUnit.MILLISECONDS);
}else{
getEntry(threadId).getLatch().acquire();
}
}
}finally{
//撤销对channel的订阅
unsubscribe(fuV n Jture D S,threadId);
}
//get(lockAsync(leaseTime,unit));
}

获取锁

获取锁的时分，也比较简略，你能够看到，他也是不断改写过期时刻，跟我上面不断去拿当时时刻，校验过期是一个道理，仅仅我比较粗糙。

private<T>RFuture<Long>tryAcquirey 2 I PAsync(loG Q ) C m & R #ngleaseTime,TimeUnitunit,finallongthreadIdk T j){

//假如带有过期时刻，则依照普通办法获取锁
if(leaseTi; K 6 3 * - 0 ume!=-1){
returntryLockInnerAsync(leaseTime,unit,threadId,RedisCommands.EVAL_LONJ r AG);
}

//先依照30秒的过期时刻来履行获取锁的办法
RFuture<Long>ttlRemainN R , 5 , }ingFuture=tryLockInnerAsync(
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS,threadId,RedisCommands.EVAL_LONG);

//假如还持有这个锁，则敞开定时任务不断改写该锁的过期时刻
ttlRemainingFuture.addList, o a H S ^ ? Y Aener(newFutureListener<Long>(){
@Override
publicvoidoperationComplete(Future<Long>futun ( K D jre)throwsException{
if(!future.isS[ t E 0 s +  cuccess()){
return;
}

LongttlRemaining=future.getNow();
//lockacquired
if(ttlRemaining==null){
scheduleExpirationRe; 0 c # w S J }newal(threadId);
}
}
});
reu a k Z 0 | vturnr 3 k 4 ) / E 9 rttlRemainingFuture;
}

底层加锁逻辑

你或许会想这么多操作，在一起不是原子性不仍是有问题么？

大佬们必定想得到呀，所以仍是LUA，他使& C ] T h k用了Hash的数据结构。

主要是判别锁是否存在，存在u P R ~ ; p a }就设置过期时刻，假如锁现已存在了，那比照一下线程，线程是一个那就证明能够重入，锁在了，可是不是当时线程，证明别人还没开释，那就把剩余时刻回来，加锁失利。

是不是有点绕，多理解一遍。

<K u ` $ = S 1 N ;;T>RFuture<T>tryLockInnerAsync(longleaseTime,TimeUi 2 7 0nitunit,
longthreadId,Rk a : L w ` ` H cedisStrictCommand<T>command)q + 1 X K y ` 4 I{

//过期时刻
internalLockL& u @ . TeaseTime=unit.toMillis(le z k / ^ #ea` ? 9  O z I m 4seTime);

rZ 9 geturnco! 2 A & S w smmandExecutor.evalWriteAsync(get! @ } y p G w IName(),LongCodec.INSTANCE,o m $ S n 0commanN 0 p 3d,
/* + D /假如锁不存在，则经过h* M (set设置它的值，并设置过期时刻
"if(x ) X C s *redis.call('exists',KEYS[1])==0)tho j , Y Qen"+
"redis.call('hset',KEYS[1],ARGV[2],1);"+
"redis.call('pe` e U O ; 8 Y Qxpire',KEYS[1],ARGV[1])k 0 & t z | } j;"+
"returnnil;"+
"end;"+
//假如锁已存在，并且锁的是当时线程，则经过hincrby给数值递加1
"if(red, U v N  w M Gis.call('hexists',KEYS[1],ARGV[2])==1), # 1then"+
"redis.call('hincrby',KEYS[1],ARGV[2],1);"+
! 7 ~ + d I X T $"redis.G B x = [ A `call('pexpire',KEYS[1]s ) * & b G M N ],ARGV[1]);"+
"returnnil;"+
"end;"+
//假如锁已存在，但并非2 P T c o 本线程，则回来过期时刻ttl
"returnredis.call('pttl',KEYS[1]);",
Collections.<Object>singletonList(getName()),
internalLockLeaseTime,getLocx R ^ 3 %  1 m jkName(threadId));
}

解锁

锁的开释主要是publish开释锁的信息，然后做校验，相同会判别是否当时线程，成功就开释锁，还有C w T u `个hincrby递减的1 Y / H : ` 2 D .操作，锁的值大于0阐明是可重入锁，那就改写过期时刻。

假如值小于0了，那删掉Key开释锁。

是不是又和AQS很像了？

AQS便是经过一个volatile修饰status去看锁的状态，也会看数值判别是否是可重入的。_ ! y r

所以我说代码的规划，最v & = N O ] a后就万剑归一，都是相同的。

publicRFuture<Void>unlockAsync(f~ 3 *inallongthreadId){
fi k ~ t m cnalRPromise<Void>result=newRedissonPromise<Void>();

//解锁办法
RFuture<Boolean>future=unlockInnerAsync(threadId);

future.addListener(newFutureListener<Boolean>(){
@Override
publicvoidoperationComplete(Future<Boolean>future)throwsException{P y e A D # h 3 M
if(L A , g L L Q } .!future.isSuccess())/ K v  5 - N U z{
cancelExpirationRenewal(threadId);g  q / r ~ $ $
resp O v o 4 Bult.tryFailure(future.cause());
rL C N Z M I U {eturn;
}
//获取回来值
BooleanopStatus=future.getNow();
//假如回来空，则证明解@ 8 0锁的线程和当时锁不是同一个线程，抛出反常
if(opStatus==null){
IllegalMonitorStateExceptioncause=
newIllegalMo9 N UnitorStateExceptio- i mn("
attempttounlocklock,notlockedbycurrentthreadbynodeid:"
+id+"thread-id:"+threadId);
result.tryFailure(cause)r r s a s;
return;
}
//J e O 7 _解锁成功，撤销改写过期时刻的那个定时任务
if(opStatus){
cancelExpirationRenewal(null);
}
result.trySuccess(null);
}
});

returnresult;
}


protecte- n U H =  q f 2dRFm b W `utc J V / , ! R [ ~ure<Boolean>unl$  mockInnerAsync(longthreadId){
returncommandExecutor.evalWriteAsync(getName(),LongCodec.INSTANCE,EO O ` k 8VAL,

//假如锁现已不存在，发布锁开释的音讯
"if(redis.call('exists',KEYS[1])==0)then"+
"redis.call('publish',KEYS[2],ARGV[1]);"+
"return1;"+
"end;"+
//假如开释锁的线程v W ~ k m 1  i b和已存在锁的线程不是同一个线程，回来null
"if(redis.call('hexists',KEYS[1],Y F r ^ iARGV[3])==0)then"+
"returnnil;"+
"end;"+
//经过hincrby递减1的办法，开释一次锁
//若剩余次数f * h ] , y大于0，则改写过期i ? A ( @时刻
"localcounter=redis.call('hincrby',KEYS[1],ARGV[3],-1);"+
"if(coun- { Y I s d ~ _ mter>0)then"+
"redis.call('pexpire',KEYS[1],ARGV[2]);"+
"return0;"+
//不然证明v ^ U _ % [锁现已开释，删去key并发布锁开释的音讯
"else"+
"redis.call('del',KEYS[1]);"+
"redis.call('pu6 k Lblish',KEYS[2],ARGV[1]);"+
"return1;"+
"end;c S R l "+
"rB c B $ = m Aetu* # ( ^ k . srnnil;",
Arrays.<ObjecF # [ @t>asList(getName(),get1 w K # % o {ChannelName()),
LockPubSub.unlockJ [ - w } n VMessage,internalLockLeaseTime,getLockName(threadId));

}

总结

这个写了比较久，可是不是由于杂乱什么的7 f J l %，是由于个人工作的原因，最近工作许多嘛，仍是那句话，程序员才是我的本职写文章仅s P g C仅个喜好，不能本末倒置了。

咱们会发现，你学懂一个技能栈之后，学新的会很快，并且也能z 2 Q B K k发现他们的规划思维和技巧真的很奇妙，也总能找到相似点，和让你惊叹的点。

就拿Dx T $ houg Lea写的AbstractQueuedSynchronizer（AQS）来说，他写了一行代码，你或许看几天才干看懂，大佬们的思维是真的牛。

我看源码有时分也头疼，可是去谷歌一下，自己理解一下，忽然茅塞顿开的时分觉得一切又很值。

学习便是一条时而闷闷不乐，时而开环大笑的路，咱们加油，咱们生长路上一起共勉。

我是敖丙，一个在互联网苟全性命的东西人。

最好的联系是互相成就，咱们的 「三连」便是丙丙创作的最大动力，咱们下期见！

注：假如本篇博客有任何错误和建议，# _ F h欢迎人才们留言，你快说句话啊！

文章继续更新，能够微信查找「 三太子敖丙 」第一时刻阅览，回复【材料】【面试】【简历) / : B p g & } n】有我预备的一线大厂面试材料和简C 3 & 8 1 B /历模板，本文 GitHub github.cok s l u F |m/JavaFamilyR v y r ^ = K i * 现已, T }录入，有大厂面试完好考点，欢迎Star。

你知道的越多，你不知道的越多