從構建分佈式秒殺系統聊聊分佈式鎖

時間 2019-11-20

原文原文鏈接

前言

最近懶成一坨屎，學不動系列一波接一波，大多還都是底層原理相關的。上週末抽時間重讀了周志明大溼的 JVM 高效併發部分，每讀一遍都有不一樣的感悟。路漫漫，藉此，把前段時間搞着玩的秒殺案例中的分佈式鎖深刻了解一下。html

案例介紹

在嘗試瞭解分佈式鎖以前，你們能夠想象一下，什麼場景下會使用分佈式鎖？算法

單機應用架構中，秒殺案例使用ReentrantLcok或者synchronized來達到秒殺商品互斥的目的。然而在分佈式系統中，會存在多臺機器並行去實現同一個功能。也就是說，在多進程中，若是還使用以上JDK提供的進程鎖，來併發訪問數據庫資源就可能會出現商品超賣的狀況。所以，須要咱們來實現本身的分佈式鎖。數據庫

實現一個分佈式鎖應該具有的特性：apache

高可用、高性能的獲取鎖與釋放鎖緩存
在分佈式系統環境下，一個方法或者變量同一時間只能被一個線程操做網絡
具有鎖失效機制，網絡中斷或宕機沒法釋放鎖時，鎖必須被刪除，防止死鎖session
具有阻塞鎖特性，即沒有獲取到鎖，則繼續等待獲取鎖架構
具有非阻塞鎖特性，即沒有獲取到鎖，則直接返回獲取鎖失敗併發
具有可重入特性，一個線程中能夠屢次獲取同一把鎖，好比一個線程在執行一個帶鎖的方法，該方法中又調用了另外一個須要相同鎖的方法，則該線程能夠直接執行調用的方法，而無需從新得到鎖分佈式

在以前的秒殺案例中，咱們曾介紹過關於分佈式鎖幾種實現方式：

基於數據庫實現分佈式鎖
基於 Redis 實現分佈式鎖
基於 Zookeeper 實現分佈式鎖

前兩種對於分佈式生產環境來講並非特別推薦，高併發下數據庫鎖性能太差，Redis在鎖時間限制和緩存一致性存在必定問題。這裏咱們重點介紹一下 Zookeeper 如何實現分佈式鎖。

實現原理

ZooKeeper是一個分佈式的，開放源碼的分佈式應用程序協調服務，它內部是一個分層的文件系統目錄樹結構，規定同一個目錄下只能存在惟一文件名。

ZooKeeper數據模型與文件系統目錄樹(源自網絡)

數據模型

PERSISTENT 持久化節點，節點建立後，不會由於會話失效而消失
EPHEMERAL 臨時節點，客戶端session超時此類節點就會被自動刪除
EPHEMERAL_SEQUENTIAL 臨時自動編號節點
PERSISTENT_SEQUENTIAL 順序自動編號持久化節點，這種節點會根據當前已存在的節點數自動加 1

監視器（watcher）

當建立一個節點時，能夠註冊一個該節點的監視器，當節點狀態發生改變時，watch被觸發時，ZooKeeper將會向客戶端發送且僅發送一條通知，由於watch只能被觸發一次。

根據zookeeper的這些特性，咱們來看看如何利用這些特性來實現分佈式鎖：

建立一個鎖目錄lock
線程A獲取鎖會在lock目錄下，建立臨時順序節點
獲取鎖目錄下全部的子節點，而後獲取比本身小的兄弟節點，若是不存在，則說明當前線程順序號最小，得到鎖
線程B建立臨時節點並獲取全部兄弟節點，判斷本身不是最小節點，設置監聽(watcher)比本身次小的節點
線程A處理完，刪除本身的節點，線程B監聽到變動事件，判斷本身是最小的節點，得到鎖

代碼分析

儘管ZooKeeper已經封裝好複雜易出錯的關鍵服務，將簡單易用的接口和性能高效、功能穩定的系統提供給用戶。可是若是讓一個普通開發者去手擼一個分佈式鎖仍是比較困難的，在秒殺案例中咱們直接使用 Apache 開源的curator 開實現 Zookeeper 分佈式鎖。

這裏咱們使用如下版本，截止目前最新版4.0.1：


<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.10.0</version>
</dependency>

首先，咱們看下InterProcessLock接口中的幾個方法：

/**
* 獲取鎖、阻塞等待、可重入
*/
public void acquire() throws Exception;
/**
* 獲取鎖、阻塞等待、可重入、超時則獲取失敗
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;
/**
* 釋放鎖
*/
public void release() throws Exception;
/**
* Returns true if the mutex is acquired by a thread in this JVM
*/
boolean isAcquiredInThisProcess();

獲取鎖：

//獲取鎖
public void acquire() throws Exception
{
if ( !internalLock(-1, null) )
{
throw new IOException("Lost connection while trying to acquire lock: " + basePath);
}
}

具體實現：

private boolean internalLock(long time, TimeUnit unit) throws Exception
{
/*
實現同一個線程可重入性，若是當前線程已經得到鎖，
則增長鎖數據中lockCount的數量(重入次數)，直接返回成功
*/
//獲取當前線程
Thread currentThread = Thread.currentThread();
//獲取當前線程重入鎖相關數據
LockData lockData = threadData.get(currentThread);
if ( lockData != null )
{
//原子遞增一個當前值，記錄重入次數，後面鎖釋放會用到
lockData.lockCount.incrementAndGet();
return true;
}
//嘗試鏈接zookeeper獲取鎖
String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
if ( lockPath != null )
{
//建立可重入鎖數據，用於記錄當前線程重入次數
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
//獲取鎖超時或者zk通訊異常返回失敗
return false;
}

Zookeeper獲取鎖實現：

String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
{
//獲取當前時間戳
final long startMillis = System.currentTimeMillis();
//若是unit不爲空(非阻塞鎖)，把當前傳入time轉爲毫秒
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
//子節點標識
final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
//嘗試次數
int retryCount = 0;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
//自旋鎖，循環獲取鎖
while ( !isDone )
{
isDone = true;
try
{
//在鎖節點下建立臨時且有序的子節點，例如:_c_008c1b07-d577-4e5f-8699-8f0f98a013b4-lock-000000001
ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
//若是當前子節點序號最小，得到鎖則直接返回，不然阻塞等待前一個子節點刪除通知(release釋放鎖)
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
}
catch ( KeeperException.NoNodeException e )
{
//異常處理，若是找不到節點，這可能發生在session過時等時，所以，若是重試容許，只需重試一次便可
if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
{
isDone = false;
}
else
{
throw e;
}
}
}
//若是獲取鎖則返回當前鎖子節點路徑
if ( hasTheLock )
{
return ourPath;
}
return null;
}

判斷是否爲最小節點：

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<String> 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)
{
try
{
//這裏使用getData()接口而不是checkExists()是由於，若是前一個子節點已經被刪除了那麼會拋出異常並且不會設置事件監聽器，而checkExists雖然也能夠獲取到節點是否存在的信息可是同時設置了監聽器，這個監聽器其實永遠不會觸發，對於Zookeeper來講屬於資源泄露
client.getData().usingWatcher(watcher).forPath(previousSequencePath);
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
//若是設置了獲取鎖等待時間
if ( millisToWait <= 0 )
{
doDelete = true; // 超時則刪除子節點
break;
}
//等待超時時間
wait(millisToWait);
}
else
{
wait();//一直等待
}
}
catch ( KeeperException.NoNodeException e )
{
// it has been deleted (i.e. lock released). Try to acquire again
//若是前一個子節點已經被刪除則deException，只須要自旋獲取一次便可
}
}
}
}
}
catch ( Exception e )
{
ThreadUtils.checkInterrupted(e);
doDelete = true;
throw e;
}
finally
{
if ( doDelete )
{
deleteOurPath(ourPath);//獲取鎖超時則刪除節點
}
}
return haveTheLock;
}

釋放鎖：

public void release() throws Exception
{
Thread currentThread = Thread.currentThread();
LockData lockData = threadData.get(currentThread);
//沒有獲取鎖，你釋放個球球，若是爲空拋出異常
if ( lockData == null )
{
throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
}
//獲取重入數量
int newLockCount = lockData.lockCount.decrementAndGet();
//若是重入鎖次數大於0，直接返回
if ( newLockCount > 0 )
{
return;
}
//若是重入鎖次數小於0，拋出異常
if ( newLockCount < 0 )
{
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
}
try
{
//釋放鎖
internals.releaseLock(lockData.lockPath);
}
finally
{
//移除當前線程鎖數據
threadData.remove(currentThread);
}
}

測試案例

爲了更好的理解其原理和代碼分析中獲取鎖的過程，這裏咱們實現一個簡單的Demo：

/**
* 基於curator的zookeeper分佈式鎖
*/
public class CuratorUtil {
private static String address = "192.168.1.180:2181";
public static void main(String[] args) {
//一、重試策略：初試時間爲1s 重試3次
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//二、經過工廠建立鏈接
CuratorFramework client = CuratorFrameworkFactory.newClient(address, retryPolicy);
//三、開啓鏈接
client.start();
//4 分佈式鎖
final InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock");
//讀寫鎖
//InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/readwriter");
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
fixedThreadPool.submit(new Runnable() {
@Override
public void run() {
boolean flag = false;
try {
//嘗試獲取鎖，最多等待5秒
flag = mutex.acquire(5, TimeUnit.SECONDS);
Thread currentThread = Thread.currentThread();
if(flag){
System.out.println("線程"+currentThread.getId()+"獲取鎖成功");
}else{
System.out.println("線程"+currentThread.getId()+"獲取鎖失敗");
}
//模擬業務邏輯，延時4秒
Thread.sleep(4000);
} catch (Exception e) {
e.printStackTrace();
} finally{
if(flag){
try {
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
});
}
}
}