大家好，我是師長

為什么用分布式鎖？

在討論這個問題之前，我們先來看一個業務場景：

系統A是一個電商系統，目前是一臺機器部署，系統中有一個用戶下訂單的接口，但是用戶下訂單之前一定要去檢查一下庫存，確保庫存足夠了才會給用戶下單。

由于系統有一定的并發，所以會預先將商品的庫存保存在redis中，用戶下單的時候會更新redis的庫存。

此時系統架構如下：

但是這樣一來會產生一個問題：假如某個時刻，redis里面的某個商品庫存為1，此時兩個請求同時到來，其中一個請求執行到上圖的第3步，更新數據庫的庫存為0，但是第4步還沒有執行。

而另外一個請求執行到了第2步，發現庫存還是1，就繼續執行第3步。

這樣的結果，是導致賣出了2個商品，然而其實庫存只有1個。

很明顯不對啊！這就是典型的庫存超賣問題

此時，我們很容易想到解決方案：用鎖把2、3、4步鎖住，讓他們執行完之后，另一個線程才能進來執行第2步。

按照上面的圖，在執行第2步時，使用Java提供的synchronized或者ReentrantLock來鎖住，然后在第4步執行完之后才釋放鎖。

這樣一來，2、3、4 這3個步驟就被“鎖”住了，多個線程之間只能串行化執行。

但是好景不長，整個系統的并發飆升，一臺機器扛不住了。現在要增加一臺機器，如下圖：

增加機器之后，系統變成上圖所示，我的天！

假設此時兩個用戶的請求同時到來，但是落在了不同的機器上，那么這兩個請求是可以同時執行了，還是會出現庫存超賣的問題。

為什么呢？因為上圖中的兩個A系統，運行在兩個不同的JVM里面，他們加的鎖只對屬于自己JVM里面的線程有效，對于其他JVM的線程是無效的。

因此，這里的問題是：Java提供的原生鎖機制在多機部署場景下失效了

這是因為兩臺機器加的鎖不是同一個鎖(兩個鎖在不同的JVM里面)。

那么，我們只要保證兩臺機器加的鎖是同一個鎖，問題不就解決了嗎？

此時，就該分布式鎖隆重登場了，分布式鎖的思路是：

在整個系統提供一個全局、唯一的獲取鎖的“東西”，然后每個系統在需要加鎖時，都去問這個“東西”拿到一把鎖，這樣不同的系統拿到的就可以認為是同一把鎖。

至于這個“東西”，可以是Redis、Zookeeper，也可以是數據庫。

文字描述不太直觀，我們來看下圖：

通過上面的分析，我們知道了庫存超賣場景在分布式部署系統的情況下使用Java原生的鎖機制無法保證線程安全，所以我們需要用到分布式鎖的方案。

那么，如何實現分布式鎖呢？接著往下看！

基于Redis實現分布式鎖

上面分析為啥要使用分布式鎖了，這里我們來具體看看分布式鎖落地的時候應該怎么樣處理。擴展：

最常見的一種方案就是使用Redis做分布式鎖

使用Redis做分布式鎖的思路大概是這樣的：在redis中設置一個值表示加了鎖，然后釋放鎖的時候就把這個key刪除。

具體代碼是這樣的：

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

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


除了要考慮客戶端要怎么實現分布式鎖之外，還需要考慮redis的部署問題。

redis有3種部署方式：

• 單機模式

• master-slave + sentinel選舉模式

• redis cluster模式

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();

就是這么簡單，我們只需要通過它的api中的lock和unlock即可完成分布式鎖，他幫我們考慮了很多細節：

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

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

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

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

redisson的“看門狗”邏輯保證了沒有死鎖發生。

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

這里稍微貼出來其實現代碼：

// 加鎖邏輯
private RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
if (leaseTime != -1) {
return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
// 調用一段lua腳本，設置一些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.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

// 看門狗最終會調用了這里
private void scheduleExpirationRenewal(final long threadId) {
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}

// 這個任務會延遲10s執行
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {

// 這個操作會將key的過期時間重新設置為30s
RFuture future = renewExpirationAsync(threadId);

future.addListener(new FutureListener() {
@Override
public void operationComplete(Future future) throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}

if (future.getNow()) {
// reschedule itself
// 通過遞歸調用本方法，無限循環延長過期時間
scheduleExpirationRenewal(threadId);
}
}
});
}

}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
task.cancel();
}
}

另外，redisson還提供了對redlock算法的支持,

它的用法也很簡單：


RedissonClient redisson = Redisson.create(config);
RLock lock1 = redisson.getFairLock("lock1");
RLock lock2 = redisson.getFairLock("lock2");
RLock lock3 = redisson.getFairLock("lock3");
RedissonRedLock multiLock = new RedissonRedLock(lock1, lock2, lock3);
multiLock.lock();
multiLock.unlock();


小結：

本節分析了使用redis作為分布式鎖的具體落地方案

以及其一些局限性

然后介紹了一個redis的客戶端框架redisson，

這也是我推薦大家使用的，

比自己寫代碼實現會少care很多細節。

基于zookeeper實現分布式鎖

常見的分布式鎖實現方案里面，除了使用redis來實現之外，使用zookeeper也可以實現分布式鎖。

在介紹zookeeper(下文用zk代替)實現分布式鎖的機制之前，先粗略介紹一下zk是什么東西：

Zookeeper是一種提供配置管理、分布式協同以及命名的中心化服務。

zk的模型是這樣的：zk包含一系列的節點，叫做znode，就好像文件系統一樣每個znode表示一個目錄，然后znode有一些特性：

有序節點：假如當前有一個父節點為/lock，我們可以在這個父節點下面創建子節點；

zookeeper提供了一個可選的有序特性，例如我們可以創建子節點“/lock/node-”并且指明有序，那么zookeeper在生成子節點時會根據當前的子節點數量自動添加整數序號

也就是說，如果是第一個創建的子節點，那么生成的子節點為/lock/node-0000000000，下一個節點則為/lock/node-0000000001，依次類推。

臨時節點：客戶端可以建立一個臨時節點，在會話結束或者會話超時后，zookeeper會自動刪除該節點。

事件監聽：在讀取數據時，我們可以同時對節點設置事件監聽，當節點數據或結構變化時，zookeeper會通知客戶端。當前zookeeper有如下四種事件：

節點創建

節點刪除

節點數據修改

子節點變更

基于以上的一些zk的特性，我們很容易得出使用zk實現分布式鎖的落地方案：

使用zk的臨時節點和有序節點，每個線程獲取鎖就是在zk創建一個臨時有序的節點，比如在/lock/目錄下。

創建節點成功后，獲取/lock目錄下的所有臨時節點，再判斷當前線程創建的節點是否是所有的節點的序號最小的節點

如果當前線程創建的節點是所有節點序號最小的節點，則認為獲取鎖成功。

如果當前線程創建的節點不是所有節點序號最小的節點，則對節點序號的前一個節點添加一個事件監聽。

比如當前線程獲取到的節點序號為/lock/003,然后所有的節點列表為[/lock/001,/lock/002,/lock/003],則對/lock/002這個節點添加一個事件監聽器。

如果鎖釋放了，會喚醒下一個序號的節點，然后重新執行第3步，判斷是否自己的節點序號是最小。

比如/lock/001釋放了，/lock/002監聽到時間，此時節點集合為[/lock/002,/lock/003],則/lock/002為最小序號節點，獲取到鎖。

整個過程如下：

具體的實現思路就是這樣，至于代碼怎么寫，這里比較復雜就不貼出來了。

Curator介紹

Curator是一個zookeeper的開源客戶端，也提供了分布式鎖的實現。

他的使用方式也比較簡單：

InterProcessMutex interProcessMutex = new InterProcessMutex(client,"/anyLock");
interProcessMutex.acquire();
interProcessMutex.release();

其實現分布式鎖的核心源碼如下：

private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
boolean haveTheLock = false;
boolean doDelete = false;
try {
if ( revocable.get() != null ) {
client.getData().usingWatcher(revocableWatcher).forPath(ourPath);


while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) {
// 獲取當前所有節點排序后的集合
List children = getSortedChildren();
// 獲取當前節點的名稱
String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
// 判斷當前節點是否是最小的節點
PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
if ( predicateResults.getsTheLock() ) {
// 獲取到鎖
haveTheLock = true;
} else {
// 沒獲取到鎖，對當前節點的上一個節點注冊一個監聽器
String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
synchronized(this){
Stat stat = client.checkExists().usingWatcher(watcher).forPath(previousSequencePath);
if ( stat != null ){
if ( millisToWait != null ){
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 ){
doDelete = true; // timed out - delete our node
break;
}
wait(millisToWait);
}else{
wait();
}
}
}
// else it may have been deleted (i.e. lock released). Try to acquire again
}
}
}
catch ( Exception e ) {
doDelete = true;
throw e;
} finally{
if ( doDelete ){
deleteOurPath(ourPath);
}
}
return haveTheLock;
}


其實curator實現分布式鎖的底層原理和上面分析的是差不多的。這里我們用一張圖詳細描述其原理：

小結：

本節介紹了zookeeperr實現分布式鎖的方案以及zk的開源客戶端的基本使用，簡要的介紹了其實現原理。相關可以參考：

兩種方案的優缺點比較

學完了兩種分布式鎖的實現方案之后，本節需要討論的是redis和zk的實現方案中各自的優缺點。

對于redis的分布式鎖而言，它有以下缺點：

它獲取鎖的方式簡單粗暴，獲取不到鎖直接不斷嘗試獲取鎖，比較消耗性能。

另外來說的話，redis的設計定位決定了它的數據并不是強一致性的，在某些極端情況下，可能會出現問題。鎖的模型不夠健壯

即便使用redlock算法來實現，在某些復雜場景下，也無法保證其實現100%沒有問題，關于redlock的討論可以看How to do distributed locking

redis分布式鎖，其實需要自己不斷去嘗試獲取鎖，比較消耗性能。

但是另一方面使用redis實現分布式鎖在很多企業中非常常見，而且大部分情況下都不會遇到所謂的“極端復雜場景”

所以使用redis作為分布式鎖也不失為一種好的方案，最重要的一點是redis的性能很高，可以支撐高并發的獲取、釋放鎖操作。

對于zk分布式鎖而言:

zookeeper天生設計定位就是分布式協調，強一致性。鎖的模型健壯、簡單易用、適合做分布式鎖。

如果獲取不到鎖，只需要添加一個監聽器就可以了，不用一直輪詢，性能消耗較小。

但是zk也有其缺點：如果有較多的客戶端頻繁的申請加鎖、釋放鎖，對于zk集群的壓力會比較大。

小結：

綜上所述，redis和zookeeper都有其優缺點。我們在做技術選型的時候可以根據這些問題作為參考因素。

建議