上一章節我提到了基于zk分布式鎖的實現，這章節就來說一下基于Redis的分布式鎖實現吧。
zk實現分布式鎖的傳送門：zk分布式鎖在開始提到Redis分布式鎖之前，我想跟大家聊點Redis的基礎知識。
說一下Redis的兩個命令：
SETNX key value
setnx 是SET if Not eXists(如果不存在，則 SET)的簡寫。
用法如圖，如果不存在set成功返回int的1，這個key存在了返回0。
SETEX key seconds value
將值 value 關聯到 key ，并將 key 的生存時間設為 seconds (以秒為單位)。
如果 key 已經存在，setex命令將覆寫舊值。
有小伙伴肯定會疑惑萬一set value 成功 set time失敗，那不就傻了么，這啊Redis官網想到了。
setex是一個原子性(atomic)操作，關聯值和設置生存時間兩個動作會在同一時間內完成。
我設置了10秒的失效時間，ttl命令可以查看倒計時，負的說明已經到期了。
跟大家講這兩個命名也是有原因的，因為他們是Redis實現分布式鎖的關鍵。
正文開始前還是看看場景:
我依然是創建了很多個線程去扣減庫存inventory，不出意外的庫存扣減順序變了，最終的結果也是不對的。
單機加synchronized或者Lock這些常規操作我就不說了好吧，結果肯定是對的。
我先實現一個簡單的Redis鎖，然后我們再實現分布式鎖，可能更方便大家的理解。
還記得上面我說過的命令么，實現一個單機的其實比較簡單，你們先思考一下，別往下看。
setnx可以看到，第一個成功了，沒釋放鎖，后面的都失敗了，至少順序問題問題是解決了，只要加鎖，縮放后面的拿到，釋放如此循環，就能保證按照順序執行。
但是你們也發現問題了，還是一樣的，第一個仔set成功了，但是突然掛了，那鎖就一直在那無法得到釋放，后面的線程也永遠得不到鎖，又死鎖了。
所以....
setex知道我之前說這個命令的原因了吧，設置一個過期時間，就算線程1掛了，也會在失效時間到了，自動釋放。
我這里就用到了nx和px的結合參數，就是set值并且加了過期時間，這里我還設置了一個過期時間，就是這時間內如果第二個沒拿到第一個的鎖，就退出阻塞了，因為可能是客戶端斷連了。
整體加鎖的邏輯比較簡單，大家基本上都能看懂，不過我拿到當前時間去減開始時間的操作感覺有點笨， System.currentTimeMillis()消耗很大的。
/**
*加鎖
*
*@paramid
*@return
*/
publicbooleanlock(Stringid){
Longstart=System.currentTimeMillis();
try{
for(;;){
//SET命令返回OK，則證明獲取鎖成功
Stringlock=jedis.set(LOCK_KEY,id,params);
if("OK".equals(lock)){
returntrue;
}
//否則循環等待，在timeout時間內仍未獲取到鎖，則獲取失敗
longl=System.currentTimeMillis()-start;
if(l>=timeout){
returnfalse;
}
try{
Thread.sleep(100);
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
}finally{
jedis.close();
}
}System.currentTimeMillis消耗大，每個線程進來都這樣，我之前寫代碼，就會在服務器啟動的時候，開一個線程不斷去拿，調用方直接獲取值就好了，不過也不是最優解，日期類還是有很多好方法的。
@Service
publicclassTimeServcie{
privatestaticlongtime;
static{
newThread(newRunnable(){
@Override
publicvoidrun(){
while(true){
try{
Thread.sleep(5);
}catch(InterruptedExceptione){
e.printStackTrace();
}
longcur=System.currentTimeMillis();
setTime(cur);
}
}
}).start();
}
publicstaticlonggetTime(){
returntime;
}
publicstaticvoidsetTime(longtime){
TimeServcie.time=time;
}
}解鎖的邏輯更加簡單，就是一段Lua的拼裝，把Key做了刪除。
你們發現沒，我上面加鎖解鎖都用了UUID，這就是為了保證，誰加鎖了誰解鎖，要是你刪掉了我的鎖，那不亂套了嘛。
LUA是原子性的，也比較簡單，就是判斷一下Key和我們參數是否相等，是的話就刪除，返回成功1，0就是失敗。
/**
*解鎖
*
*@paramid
*@return
*/
publicbooleanunlock(Stringid){
Stringscript=
"ifredis.call('get',KEYS[1])==ARGV[1]then"+
"returnredis.call('del',KEYS[1])"+
"else"+
"return0"+
"end";
try{
Stringresult=jedis.eval(script,Collections.singletonList(LOCK_KEY),Collections.singletonList(id)).toString();
return"1".equals(result)?true:false;
}finally{
jedis.close();
}
}我們可以用我們寫的Redis鎖試試效果，可以看到都按照順序去執行了
大家是不是覺得完美了，但是上面的鎖，有不少瑕疵的，我沒思考很多點，你或許可以思考一下，源碼我都開源到我的GItHub了。
而且，鎖一般都是需要可重入行的，上面的線程都是執行完了就釋放了，無法再次進入了，進去也是重新加鎖了，對于一個鎖的設計來說肯定不是很合理的。
我不打算手寫，因為都有現成的，別人幫我們寫好了。
redissonredisson的鎖，就實現了可重入了，但是他的源碼比較晦澀難懂。
使用起來很簡單，因為他們底層都封裝好了，你連接上你的Redis客戶端，他幫你做了我上面寫的一切，然后更完美。
簡單看看他的使用吧，跟正常使用Lock沒啥區別。
ThreadPoolExecutorthreadPoolExecutor=
newThreadPoolExecutor(inventory,inventory,10L,SECONDS,linkedBlockingQueue);
longstart=System.currentTimeMillis();
Configconfig=newConfig();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
finalRedissonClientclient=Redisson.create(config);
finalRLocklock=client.getLock("lock1");
for(inti=0;i<=NUM;i++){
threadPoolExecutor.execute(newRunnable(){
publicvoidrun(){
lock.lock();
inventory--;
System.out.println(inventory);
lock.unlock();
}
});
}
longend=System.currentTimeMillis();
System.out.println("執行線程數:"+NUM+"總耗時:"+(end-start)+"庫存數為:"+inventory);上面可以看到我用到了getLock，其實就是獲取一個鎖的實例。
RedissionLock也沒做啥，就是熟悉的初始化。
publicRLockgetLock(Stringname){
returnnewRedissonLock(connectionManager.getCommandExecutor(),name);
}
publicRedissonLock(CommandAsyncExecutorcommandExecutor,Stringname){
super(commandExecutor,name);
//命令執行器
this.commandExecutor=commandExecutor;
//UUID字符串
this.id=commandExecutor.getConnectionManager().getId();
//內部鎖過期時間
this.internalLockLeaseTime=commandExecutor.
getConnectionManager().getCfg().getLockWatchdogTimeout();
this.entryName=id+":"+name;
}有沒有發現很多跟Lock很多相似的地方呢？
嘗試加鎖，拿到當前線程，然后我開頭說的ttl也看到了，是不是一切都是那么熟悉？
publicvoidlockInterruptibly(longleaseTime,TimeUnitunit)throwsInterruptedException{
//當前線程ID
longthreadId=Thread.currentThread().getId();
//嘗試獲取鎖
Longttl=tryAcquire(leaseTime,unit,threadId);
//如果ttl為空，則證明獲取鎖成功
if(ttl==null){
return;
}
//如果獲取鎖失敗，則訂閱到對應這個鎖的channel
RFuture<RedissonLockEntry>future=subscribe(threadId);
commandExecutor.syncSubscription(future);
try{
while(true){
//再次嘗試獲取鎖
ttl=tryAcquire(leaseTime,unit,threadId);
//ttl為空，說明成功獲取鎖，返回
if(ttl==null){
break;
}
//ttl大于0則等待ttl時間后繼續嘗試獲取
if(ttl>=0){
getEntry(threadId).getLatch().tryAcquire(ttl,TimeUnit.MILLISECONDS);
}else{
getEntry(threadId).getLatch().acquire();
}
}
}finally{
//取消對channel的訂閱
unsubscribe(future,threadId);
}
//get(lockAsync(leaseTime,unit));
}獲取鎖的時候，也比較簡單，你可以看到，他也是不斷刷新過期時間，跟我上面不斷去拿當前時間，校驗過期是一個道理，只是我比較粗糙。
private<T>RFuture<Long>tryAcquireAsync(longleaseTime,TimeUnitunit,finallongthreadId){
//如果帶有過期時間，則按照普通方式獲取鎖
if(leaseTime!=-1){
returntryLockInnerAsync(leaseTime,unit,threadId,RedisCommands.EVAL_LONG);
}
//先按照30秒的過期時間來執行獲取鎖的方法
RFuture<Long>ttlRemainingFuture=tryLockInnerAsync(
commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
TimeUnit.MILLISECONDS,threadId,RedisCommands.EVAL_LONG);
//如果還持有這個鎖，則開啟定時任務不斷刷新該鎖的過期時間
ttlRemainingFuture.addListener(newFutureListener<Long>(){
@Override
publicvoidoperationComplete(Future<Long>future)throwsException{
if(!future.isSuccess()){
return;
}
LongttlRemaining=future.getNow();
//lockacquired
if(ttlRemaining==null){
scheduleExpirationRenewal(threadId);
}
}
});
returnttlRemainingFuture;
}你可能會想這么多操作，在一起不是原子性不還是有問題么？
大佬們肯定想得到呀，所以還是LUA，他使用了Hash的數據結構。
主要是判斷鎖是否存在，存在就設置過期時間，如果鎖已經存在了，那對比一下線程，線程是一個那就證明可以重入，鎖在了，但是不是當前線程，證明別人還沒釋放，那就把剩余時間返回，加鎖失敗。
是不是有點繞，多理解一遍。
<T>RFuture<T>tryLockInnerAsync(longleaseTime,TimeUnitunit,
longthreadId,RedisStrictCommand<T>command){
//過期時間
internalLockLeaseTime=unit.toMillis(leaseTime);
returncommandExecutor.evalWriteAsync(getName(),LongCodec.INSTANCE,command,
//如果鎖不存在，則通過hset設置它的值，并設置過期時間
"if(redis.call('exists',KEYS[1])==0)then"+
"redis.call('hset',KEYS[1],ARGV[2],1);"+
"redis.call('pexpire',KEYS[1],ARGV[1]);"+
"returnnil;"+
"end;"+
//如果鎖已存在，并且鎖的是當前線程，則通過hincrby給數值遞增1
"if(redis.call('hexists',KEYS[1],ARGV[2])==1)then"+
"redis.call('hincrby',KEYS[1],ARGV[2],1);"+
"redis.call('pexpire',KEYS[1],ARGV[1]);"+
"returnnil;"+
"end;"+
//如果鎖已存在，但并非本線程，則返回過期時間ttl
"returnredis.call('pttl',KEYS[1]);",
Collections.<Object>singletonList(getName()),
internalLockLeaseTime,getLockName(threadId));
}鎖的釋放主要是publish釋放鎖的信息，然后做校驗，一樣會判斷是否當前線程，成功就釋放鎖，還有個hincrby遞減的操作，鎖的值大于0說明是可重入鎖，那就刷新過期時間。
如果值小于0了，那刪掉Key釋放鎖。
是不是又和AQS很像了？
AQS就是通過一個volatile修飾status去看鎖的狀態，也會看數值判斷是否是可重入的。
所以我說代碼的設計，最后就萬劍歸一，都是一樣的。
publicRFuture<Void>unlockAsync(finallongthreadId){
finalRPromise<Void>result=newRedissonPromise<Void>();
//解鎖方法
RFuture<Boolean>future=unlockInnerAsync(threadId);
future.addListener(newFutureListener<Boolean>(){
@Override
publicvoidoperationComplete(Future<Boolean>future)throwsException{
if(!future.isSuccess()){
cancelExpirationRenewal(threadId);
result.tryFailure(future.cause());
return;
}
//獲取返回值
BooleanopStatus=future.getNow();
//如果返回空，則證明解鎖的線程和當前鎖不是同一個線程，拋出異常
if(opStatus==null){
IllegalMonitorStateExceptioncause=
newIllegalMonitorStateException("
attempttounlocklock,notlockedbycurrentthreadbynodeid:"
+id+"thread-id:"+threadId);
result.tryFailure(cause);
return;
}
//解鎖成功，取消刷新過期時間的那個定時任務
if(opStatus){
cancelExpirationRenewal(null);
}
result.trySuccess(null);
}
});
returnresult;
}
protectedRFuture<Boolean>unlockInnerAsync(longthreadId){
returncommandExecutor.evalWriteAsync(getName(),LongCodec.INSTANCE,EVAL,
//如果鎖已經不存在，發布鎖釋放的消息
"if(redis.call('exists',KEYS[1])==0)then"+
"redis.call('publish',KEYS[2],ARGV[1]);"+
"return1;"+
"end;"+
//如果釋放鎖的線程和已存在鎖的線程不是同一個線程，返回null
"if(redis.call('hexists',KEYS[1],ARGV[3])==0)then"+
"returnnil;"+
"end;"+
//通過hincrby遞減1的方式，釋放一次鎖
//若剩余次數大于0，則刷新過期時間
"localcounter=redis.call('hincrby',KEYS[1],ARGV[3],-1);"+
"if(counter>0)then"+
"redis.call('pexpire',KEYS[1],ARGV[2]);"+
"return0;"+
//否則證明鎖已經釋放，刪除key并發布鎖釋放的消息
"else"+
"redis.call('del',KEYS[1]);"+
"redis.call('publish',KEYS[2],ARGV[1]);"+
"return1;"+
"end;"+
"returnnil;",
Arrays.<Object>asList(getName(),getChannelName()),
LockPubSub.unlockMessage,internalLockLeaseTime,getLockName(threadId));
}這個寫了比較久，但是不是因為復雜什么的，是因為個人工作的原因，最近事情很多嘛，還是那句話，程序員才是我的本職寫文章只是個愛好，不能本末倒置了。
大家會發現，你學懂一個技術棧之后，學新的會很快，而且也能發現他們的設計思想和技巧真的很巧妙，也總能找到相似點，和讓你驚嘆的點。
就拿Doug Lea寫的AbstractQueuedSynchronizer（AQS）來說，他寫了一行代碼，你可能看幾天才能看懂，大佬們的思想是真的牛。
我看源碼有時候也頭疼，但是去谷歌一下，自己理解一下，突然恍然大悟的時候覺得一切又很值。
學習就是一條時而郁郁寡歡，時而開環大笑的路，大家加油，我們成長路上一起共勉。
我是敖丙，一個在互聯網茍且偷生的工具人。
最好的關系是互相成就，大家的**「三連」**就是丙丙創作的最大動力，我們下期見！
注：如果本篇博客有任何錯誤和建議，歡迎人才們留言，你快說句話啊！
你知道的越多，你不知道的越多
