前言
最近懶成一坨屎，學不動系列一波接一波，大多還都是底層原理相關的。上周末抽時間重讀了周志明大濕的 JVM 高效并發(fā)部分，每讀一遍都有不同的感悟。路漫漫，借此，把前段時間搞著玩的秒殺案例中的分布式鎖深入了解一下。
案例介紹
在嘗試了解分布式鎖之前，大家可以想象一下，什么場景下會使用分布式鎖？
單機應用架構中，秒殺案例使用ReentrantLcok或者synchronized來達到秒殺商品互斥的目的。然而在分布式系統(tǒng)中，會存在多臺機器并行去實現(xiàn)同一個功能。也就是說，在多進程中，如果還使用以上JDK提供的進程鎖，來并發(fā)訪問數(shù)據(jù)庫資源就可能會出現(xiàn)商品超賣的情況。因此，需要我們來實現(xiàn)自己的分布式鎖。
實現(xiàn)一個分布式鎖應該具備的特性：
高可用、高性能的獲取鎖與釋放鎖在分布式系統(tǒng)環(huán)境下，一個方法或者變量同一時間只能被一個線程操作具備鎖失效機制，網(wǎng)絡中斷或宕機無法釋放鎖時，鎖必須被刪除，防止死鎖具備阻塞鎖特性，即沒有獲取到鎖，則繼續(xù)等待獲取鎖具備非阻塞鎖特性，即沒有獲取到鎖，則直接返回獲取鎖失敗具備可重入特性，一個線程中可以多次獲取同一把鎖，比如一個線程在執(zhí)行一個帶鎖的方法，該方法中又調用了另一個需要相同鎖的方法，則該線程可以直接執(zhí)行調用的方法，而無需重新獲得鎖在之前的秒殺案例中，我們曾介紹過關于分布式鎖幾種實現(xiàn)方式：
基于數(shù)據(jù)庫實現(xiàn)分布式鎖基于 Redis 實現(xiàn)分布式鎖基于 Zookeeper 實現(xiàn)分布式鎖前兩種對于分布式生產(chǎn)環(huán)境來說并不是特別推薦，高并發(fā)下數(shù)據(jù)庫鎖性能太差，Redis在鎖時間限制和緩存一致性存在一定問題。這里我們重點介紹一下 Zookeeper 如何實現(xiàn)分布式鎖。
實現(xiàn)原理
ZooKeeper是一個分布式的，開放源碼的分布式應用程序協(xié)調服務，它內部是一個分層的文件系統(tǒng)目錄樹結構，規(guī)定同一個目錄下只能存在唯一文件名。
數(shù)據(jù)模型
PERSISTENT 持久化節(jié)點，節(jié)點創(chuàng)建后，不會因為會話失效而消失EPHEMERAL 臨時節(jié)點， 客戶端session超時此類節(jié)點就會被自動刪除EPHEMERAL_SEQUENTIAL 臨時自動編號節(jié)點PERSISTENT_SEQUENTIAL 順序自動編號持久化節(jié)點，這種節(jié)點會根據(jù)當前已存在的節(jié)點數(shù)自動加 1監(jiān)視器（watcher）
當創(chuàng)建一個節(jié)點時，可以注冊一個該節(jié)點的監(jiān)視器，當節(jié)點狀態(tài)發(fā)生改變時，watch被觸發(fā)時，ZooKeeper將會向客戶端發(fā)送且僅發(fā)送一條通知，因為watch只能被觸發(fā)一次。
根據(jù)zookeeper的這些特性，我們來看看如何利用這些特性來實現(xiàn)分布式鎖：
創(chuàng)建一個鎖目錄lock線程A獲取鎖會在lock目錄下，創(chuàng)建臨時順序節(jié)點獲取鎖目錄下所有的子節(jié)點，然后獲取比自己小的兄弟節(jié)點，如果不存在，則說明當前線程順序號最小，獲得鎖線程B創(chuàng)建臨時節(jié)點并獲取所有兄弟節(jié)點，判斷自己不是最小節(jié)點，設置監(jiān)聽(watcher)比自己次小的節(jié)點（只關注比自己次小的節(jié)點是為了防止發(fā)生“羊群效應”）線程A處理完，刪除自己的節(jié)點，線程B監(jiān)聽到變更事件，判斷自己是最小的節(jié)點，獲得鎖代碼分析
盡管ZooKeeper已經(jīng)封裝好復雜易出錯的關鍵服務，將簡單易用的接口和性能高效、功能穩(wěn)定的系統(tǒng)提供給用戶。但是如果讓一個普通開發(fā)者去手擼一個分布式鎖還是比較困難的，在秒殺案例中我們直接使用 Apache 開源的curator 開實現(xiàn) Zookeeper 分布式鎖。
這里我們使用以下版本，截止目前最新版4.0.1：
<!-- zookeeper 分布式鎖、注意zookeeper版本 這里對應的是3.4.6-->
<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
{
/*
實現(xiàn)同一個線程可重入性，如果當前線程已經(jīng)獲得鎖，
則增加鎖數(shù)據(jù)中l(wèi)ockCount的數(shù)量(重入次數(shù))，直接返回成功
*/
//獲取當前線程
Thread currentThread = Thread.currentThread();
//獲取當前線程重入鎖相關數(shù)據(jù)
LockData lockData = threadData.get(currentThread);
if ( lockData != null )
{
//原子遞增一個當前值，記錄重入次數(shù)，后面鎖釋放會用到
lockData.lockCount.incrementAndGet();
return true;
}
//嘗試連接zookeeper獲取鎖
String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
if ( lockPath != null )
{
//創(chuàng)建可重入鎖數(shù)據(jù)，用于記錄當前線程重入次數(shù)
LockData newLockData = new LockData(currentThread, lockPath);
threadData.put(currentThread, newLockData);
return true;
}
//獲取鎖超時或者zk通信異常返回失敗
return false;
}
Zookeeper獲取鎖實現(xiàn)：
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;
//子節(jié)點標識
final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
//嘗試次數(shù)
int retryCount = 0;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
//自旋鎖，循環(huán)獲取鎖
while ( !isDone )
{
isDone = true;
try
{
//在鎖節(jié)點下創(chuàng)建臨時且有序的子節(jié)點，例如
:_c_008c1b07-d577-4e5f-8699-8f0f98a013b4-lock-000000001ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
//如果當前子節(jié)點序號最小，獲得鎖則直接返回，否則阻塞等待前一個子節(jié)點刪除通知(release釋放鎖)
hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
}
catch ( KeeperException.NoNodeException e )
{
//異常處理，如果找不到節(jié)點，這可能發(fā)生在session過期等時，因此，如果重試允許，只需重試一次即可
if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
{
isDone = false;
}
else
{
throw e;
}
}
}
//如果獲取鎖則返回當前鎖子節(jié)點路徑
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 )
{
//獲取所有子節(jié)點集合
List<String> children = getSortedChildren();
//判斷當前子節(jié)點是否為最小子節(jié)點
String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
//如果是最小節(jié)點則獲取鎖
if ( predicateResults.getsTheLock() )
{
haveTheLock = true;
}
else
{
//獲取前一個節(jié)點，用于監(jiān)聽
String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
synchronized(this)
{
try
{
//這里使用getData()接口而不是checkExists()是因為，如果前一個子節(jié)點已經(jīng)被刪除了那么會拋出異常而且不會設置事件監(jiān)聽器，而checkExists雖然也可以獲取到節(jié)點是否存在的信息但是同時設置了監(jiān)聽器，這個監(jiān)聽器其實永遠不會觸發(fā)，對于Zookeeper來說屬于資源泄露
client.getData().usingWatcher(watcher).forPath(previousSequencePath);
if ( millisToWait != null )
{
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
//如果設置了獲取鎖等待時間
if ( millisToWait <= 0 )
{
doDelete = true; // 超時則刪除子節(jié)點
break;
}
//等待超時時間
wait(millisToWait);
}
else
{
wait();//一直等待
}
}
catch ( KeeperException.NoNodeException e )
{
// it has been deleted (i.e. lock released). Try to acquire again
//如果前一個子節(jié)點已經(jīng)被刪除則deException，只需要自旋獲取一次即可
}
}
}
}
}
catch ( Exception e )
{
ThreadUtils.checkInterrupted(e);
doDelete = true;
throw e;
}
finally
{
if ( doDelete )
{
deleteOurPath(ourPath);//獲取鎖超時則刪除節(jié)點
}
}
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);
}
//獲取重入數(shù)量
int newLockCount = lockData.lockCount.decrementAndGet();
//如果重入鎖次數(shù)大于0，直接返回
if ( newLockCount > 0 )
{
return;
}
//如果重入鎖次數(shù)小于0，拋出異常
if ( newLockCount < 0 )
{
throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
}
try
{
//釋放鎖
internals.releaseLock(lockData.lockPath);
}
finally
{
//移除當前線程鎖數(shù)據(jù)
threadData.remove(currentThread);
}
}
測試案例
為了更好的理解其原理和代碼分析中獲取鎖的過程，這里我們實現(xiàn)一個簡單的Demo：
/**
* 基于curator的zookeeper分布式鎖
*/
public class CuratorUtil {
private static String address = "192.168.1.180:2181";
public static void main(String[] args) {
//1、重試策略：初試時間為1s 重試3次
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
//2、通過工廠創(chuàng)建連接
CuratorFramework client = CuratorFrameworkFactory.newClient(address, retryPolicy);
//3、開啟連接
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()+"獲取鎖失敗");
}
//模擬業(yè)務邏輯，延時4秒
Thread.sleep(4000);
} catch (Exception e) {
e.printStackTrace();
} finally{
if(flag){
try {
mutex.release();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
});
}
}
}
這里我們開啟5個線程，每個線程獲取鎖的最大等待時間為5秒，為了模擬具體業(yè)務場景，方法中設置4秒等待時間。開始執(zhí)行main方法，通過ZooInspector監(jiān)控/curator/lock下的節(jié)點如下圖：
對，沒錯，設置4秒的業(yè)務處理時長就是為了觀察生成了幾個順序節(jié)點。果然如案例中所述，每個線程都會生成一個節(jié)點并且還是有序的。
觀察控制臺，我們會發(fā)現(xiàn)只有兩個線程獲取鎖成功，另外三個線程超時獲取鎖失敗會自動刪除節(jié)點。線程執(zhí)行完畢我們刷新一下/curator/lock節(jié)點，發(fā)現(xiàn)剛才創(chuàng)建的五個子節(jié)點已經(jīng)不存在了。
小結:
通過分析第三方開源工具實現(xiàn)的分布式鎖方式，收獲還是滿滿的。學習本身就是一個由淺入深的過程，從如何調用API，到理解其代碼邏輯實現(xiàn)，想要更深入可以去挖掘Zookeeper的核心算法ZAB協(xié)議。
最后為了方便大家學習，總結了學習過程中遇到的幾個關鍵詞：重入鎖、自旋鎖、有序節(jié)點、阻塞、非阻塞、監(jiān)聽，希望對大家有所幫助。
對Java微服務、分布式、高并發(fā)、高可用、大型互聯(lián)網(wǎng)架構技術、面試經(jīng)驗交流感興趣的?？梢躁P注我的頭條號，我會在微頭條不定期的發(fā)放免費的資料鏈接，這些資料都是從各個技術網(wǎng)站搜集、整理出來的，如果你有好的學習資料可以私聊發(fā)我，我會注明出處之后分享給大家。歡迎分享，歡迎評論，歡迎轉發(fā)！
以上就是【不敢相信！快來看（多少并發(fā)量需要分布式）分布式存儲時延-從構建分布式秒殺系統(tǒng)聊聊分布式鎖】的全部內容。
評論