分布式鎖用 Redis 還是 Zookeeper?

作者：jianfeng來源：石杉的架構(gòu)筆記

為什么用分布式鎖？

基于Redis實現(xiàn)分布式鎖

//?獲取鎖
//?NX是指如果key不存在就成功，key存在返回false，PX可以指定過期時間
SET?anyLock?unique_value?NX?PX?30000


//?釋放鎖：通過執(zhí)行一段lua腳本
//?釋放鎖涉及到兩條指令，這兩條指令不是原子性的
//?需要用到redis的lua腳本支持特性，redis執(zhí)行l(wèi)ua腳本是原子性的
if?redis.call(get,KEYS[1])?==?ARGV[1]?then
return?redis.call(del,KEYS[1])
else
return?0
end

• 一定要用SET key value NX PX milliseconds 命令

  如果不用，先設(shè)置了值，再設(shè)置過期時間，這個不是原子性操作，有可能在設(shè)置過期時間之前宕機，會造成死鎖(key永久存在)

• value要具有唯一性

  這個是為了在解鎖的時候，需要驗證value是和加鎖的一致才刪除key。

  這是避免了一種情況：假設(shè)A獲取了鎖，過期時間30s，此時35s之后，鎖已經(jīng)自動釋放了，A去釋放鎖，但是此時可能B獲取了鎖。A客戶端就不能刪除B的鎖了。

• 單機模式

• master-slave + sentinel選舉模式

• redis cluster模式

• 獲取當(dāng)前時間戳，單位是毫秒

• 輪流嘗試在每個master節(jié)點上創(chuàng)建鎖，過期時間設(shè)置較短，一般就幾十毫秒

• 嘗試在大多數(shù)節(jié)點上建立一個鎖，比如5個節(jié)點就要求是3個節(jié)點（n / 2 +1）

• 客戶端計算建立好鎖的時間，如果建立鎖的時間小于超時時間，就算建立成功了

• 要是鎖建立失敗了，那么就依次刪除這個鎖

• 只要別人建立了一把分布式鎖，你就得不斷輪詢?nèi)L試獲取鎖

另一種方式：Redisson

• SET anyLock unique_value NX PX 30000

Config?config?=?new?Config();
config.useClusterServers()
.addNodeAddress(redis://192.168.31.101:7001)
.addNodeAddress(redis://192.168.31.101:7002)
.addNodeAddress(redis://192.168.31.101:7003)
.addNodeAddress(redis://192.168.31.102:7001)
.addNodeAddress(redis://192.168.31.102:7002)
.addNodeAddress(redis://192.168.31.102:7003);

RedissonClient?redisson?=?Redisson.create(config);


RLock?lock?=?redisson.getLock(anyLock);
lock.lock();
lock.unlock();

• redisson所有指令都通過lua腳本執(zhí)行，redis支持lua腳本原子性執(zhí)行

• redisson設(shè)置一個key的默認(rèn)過期時間為30s,如果某個客戶端持有一個鎖超過了30s怎么辦？

  redisson中有一個watchdog的概念，翻譯過來就是看門狗，它會在你獲取鎖之后，每隔10秒幫你把key的超時時間設(shè)為30s

  這樣的話，就算一直持有鎖也不會出現(xiàn)key過期了，其他線程獲取到鎖的問題了。

• redisson的“看門狗”邏輯保證了沒有死鎖發(fā)生。

  (如果機器宕機了，看門狗也就沒了。此時就不會延長key的過期時間，到了30s之后就會自動過期了，其他線程可以獲取到鎖)

//?加鎖邏輯
private??RFuture?tryAcquireAsync(long?leaseTime,?TimeUnit?unit,?final?long?threadId)?{
????if?(leaseTime?!=?-1)?{
????????return?tryLockInnerAsync(leaseTime,?unit,?threadId,?RedisCommands.EVAL_LONG);
????}
????//?調(diào)用一段lua腳本，設(shè)置一些key、過期時間
????RFuture?ttlRemainingFuture?=?tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),?TimeUnit.MILLISECONDS,?threadId,?RedisCommands.EVAL_LONG);
????ttlRemainingFuture.addListener(new?FutureListener()?{
????????@Override
????????public?void?operationComplete(Future?future)?throws?Exception?{
????????????if?(!future.isSuccess())?{
????????????????return;
????????????}

????????????Long?ttlRemaining?=?future.getNow();
????????????//?lock?acquired
????????????if?(ttlRemaining?==?null)?{
????????????????//?看門狗邏輯
????????????????scheduleExpirationRenewal(threadId);
????????????}
????????}
????});
????return?ttlRemainingFuture;
}


?RFuture?tryLockInnerAsync(long?leaseTime,?TimeUnit?unit,?long?threadId,?RedisStrictCommand?command)?{
????internalLockLeaseTime?=?unit.toMillis(leaseTime);

????return?commandExecutor.evalWriteAsync(getName(),?LongCodec.INSTANCE,?command,
??????????????if?(redis.call('exists',?KEYS[1])?==?0)?then??+
??????????????????redis.call('hset',?KEYS[1],?ARGV[2],?1);??+
??????????????????redis.call('pexpire',?KEYS[1],?ARGV[1]);??+
??????????????????return?nil;??+
??????????????end;??+
??????????????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]);??+
??????????????????return?nil;??+
??????????????end;??+
??????????????return?redis.call('pttl',?KEYS[1]);,
????????????????Collections.