您好,登錄后才能下訂單哦!
深入淺析ReentrantL中實(shí)現(xiàn)ock的原理?相信很多沒(méi)有經(jīng)驗(yàn)的人對(duì)此束手無(wú)策,為此本文總結(jié)了問(wèn)題出現(xiàn)的原因和解決方法,通過(guò)這篇文章希望你能解決這個(gè)問(wèn)題。
1 synchronized和lock
1.1 synchronized的局限性
1.2 Lock簡(jiǎn)介
2 AQS
3 lock()與unlock()實(shí)現(xiàn)原理
3.1 基礎(chǔ)知識(shí)
3.2 內(nèi)部結(jié)構(gòu)
3.3 NonfairSync
3.3.1 lock()
3.3.2 unlock()
3.3.3 小結(jié)
3.4 FairSync
4 超時(shí)機(jī)制
5 總結(jié)
1 synchronized和lock
1.1 synchronized的局限性
synchronized是java內(nèi)置的關(guān)鍵字,它提供了一種獨(dú)占的加鎖方式。synchronized的獲取和釋放鎖由JVM實(shí)現(xiàn),用戶(hù)不需要顯示的釋放鎖,非常方便。然而synchronized也有一定的局限性,例如:
當(dāng)線程嘗試獲取鎖的時(shí)候,如果獲取不到鎖會(huì)一直阻塞。
如果獲取鎖的線程進(jìn)入休眠或者阻塞,除非當(dāng)前線程異常,否則其他線程嘗試獲取鎖必須一直等待。
JDK1.5之后發(fā)布,加入了Doug Lea實(shí)現(xiàn)的concurrent包。包內(nèi)提供了Lock類(lèi),用來(lái)提供更多擴(kuò)展的加鎖功能。Lock彌補(bǔ)了synchronized的局限,提供了更加細(xì)粒度的加鎖功能。
1.2 Lock簡(jiǎn)介
Lock api如下
void lock(); void lockInterruptibly() throws InterruptedException; boolean tryLock(); boolean tryLock(long time, TimeUnit unit) throws InterruptedException; void unlock(); Condition newCondition();
其中最常用的就是lock和unlock操作了。因?yàn)槭褂胠ock時(shí),需要手動(dòng)的釋放鎖,所以需要使用try..catch來(lái)包住業(yè)務(wù)代碼,并且在finally中釋放鎖。典型使用如下
private Lock lock = new ReentrantLock(); public void test(){ lock.lock(); try{ doSomeThing(); }catch (Exception e){ // ignored }finally { lock.unlock(); } }
2 AQS
AbstractQueuedSynchronizer簡(jiǎn)稱(chēng)AQS,是一個(gè)用于構(gòu)建鎖和同步容器的框架。事實(shí)上concurrent包內(nèi)許多類(lèi)都是基于AQS構(gòu)建,例如ReentrantLock,Semaphore,CountDownLatch,ReentrantReadWriteLock,F(xiàn)utureTask等。AQS解決了在實(shí)現(xiàn)同步容器時(shí)設(shè)計(jì)的大量細(xì)節(jié)問(wèn)題。
AQS使用一個(gè)FIFO的隊(duì)列表示排隊(duì)等待鎖的線程,隊(duì)列頭節(jié)點(diǎn)稱(chēng)作“哨兵節(jié)點(diǎn)”或者“啞節(jié)點(diǎn)”,它不與任何線程關(guān)聯(lián)。其他的節(jié)點(diǎn)與等待線程關(guān)聯(lián),每個(gè)節(jié)點(diǎn)維護(hù)一個(gè)等待狀態(tài)waitStatus。如圖
AQS中還有一個(gè)表示狀態(tài)的字段state,例如ReentrantLocky用它表示線程重入鎖的次數(shù),Semaphore用它表示剩余的許可數(shù)量,F(xiàn)utureTask用它表示任務(wù)的狀態(tài)。對(duì)state變量值的更新都采用CAS操作保證更新操作的原子性。
AbstractQueuedSynchronizer繼承了AbstractOwnableSynchronizer,這個(gè)類(lèi)只有一個(gè)變量:exclusiveOwnerThread,表示當(dāng)前占用該鎖的線程,并且提供了相應(yīng)的get,set方法。
理解AQS可以幫助我們更好的理解JCU包中的同步容器。
3 lock()與unlock()實(shí)現(xiàn)原理
3.1 基礎(chǔ)知識(shí)
ReentrantLock是Lock的默認(rèn)實(shí)現(xiàn)之一。那么lock()和unlock()是怎么實(shí)現(xiàn)的呢?首先我們要弄清楚幾個(gè)概念
可重入鎖。可重入鎖是指同一個(gè)線程可以多次獲取同一把鎖。ReentrantLock和synchronized都是可重入鎖。
可中斷鎖。可中斷鎖是指線程嘗試獲取鎖的過(guò)程中,是否可以響應(yīng)中斷。synchronized是不可中斷鎖,而ReentrantLock則提供了中斷功能。
公平鎖與非公平鎖。公平鎖是指多個(gè)線程同時(shí)嘗試獲取同一把鎖時(shí),獲取鎖的順序按照線程達(dá)到的順序,而非公平鎖則允許線程“插隊(duì)”。synchronized是非公平鎖,而ReentrantLock的默認(rèn)實(shí)現(xiàn)是非公平鎖,但是也可以設(shè)置為公平鎖。
CAS操作(CompareAndSwap)。CAS操作簡(jiǎn)單的說(shuō)就是比較并交換。CAS 操作包含三個(gè)操作數(shù) —— 內(nèi)存位置(V)、預(yù)期原值(A)和新值(B)。如果內(nèi)存位置的值與預(yù)期原值相匹配,那么處理器會(huì)自動(dòng)將該位置值更新為新值。否則,處理器不做任何操作。無(wú)論哪種情況,它都會(huì)在 CAS 指令之前返回該位置的值。CAS 有效地說(shuō)明了“我認(rèn)為位置 V 應(yīng)該包含值 A;如果包含該值,則將 B 放到這個(gè)位置;否則,不要更改該位置,只告訴我這個(gè)位置現(xiàn)在的值即可。” Java并發(fā)包(java.util.concurrent)中大量使用了CAS操作,涉及到并發(fā)的地方都調(diào)用了sun.misc.Unsafe類(lèi)方法進(jìn)行CAS操作。
3.2 內(nèi)部結(jié)構(gòu)
ReentrantLock提供了兩個(gè)構(gòu)造器,分別是
public ReentrantLock() { sync = new NonfairSync(); } public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
默認(rèn)構(gòu)造器初始化為NonfairSync對(duì)象,即非公平鎖,而帶參數(shù)的構(gòu)造器可以指定使用公平鎖和非公平鎖。由lock()和unlock的源碼可以看到,它們只是分別調(diào)用了sync對(duì)象的lock()和release(1)方法。
Sync是ReentrantLock的內(nèi)部類(lèi),它的結(jié)構(gòu)如下
可以看到Sync擴(kuò)展了AbstractQueuedSynchronizer。
3.3 NonfairSync
我們從源代碼出發(fā),分析非公平鎖獲取鎖和釋放鎖的過(guò)程。
3.3.1 lock()
lock()源碼如下
final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
首先用一個(gè)CAS操作,判斷state是否是0(表示當(dāng)前鎖未被占用),如果是0則把它置為1,并且設(shè)置當(dāng)前線程為該鎖的獨(dú)占線程,表示獲取鎖成功。當(dāng)多個(gè)線程同時(shí)嘗試占用同一個(gè)鎖時(shí),CAS操作只能保證一個(gè)線程操作成功,剩下的只能乖乖的去排隊(duì)啦。
“非公平”即體現(xiàn)在這里,如果占用鎖的線程剛釋放鎖,state置為0,而排隊(duì)等待鎖的線程還未喚醒時(shí),新來(lái)的線程就直接搶占了該鎖,那么就“插隊(duì)”了。
若當(dāng)前有三個(gè)線程去競(jìng)爭(zhēng)鎖,假設(shè)線程A的CAS操作成功了,拿到了鎖開(kāi)開(kāi)心心的返回了,那么線程B和C則設(shè)置state失敗,走到了else里面。我們往下看acquire。
acquire(arg)
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
代碼非常簡(jiǎn)潔,但是背后的邏輯卻非常復(fù)雜,可見(jiàn)Doug Lea大神的編程功力。
1. 第一步。嘗試去獲取鎖。如果嘗試獲取鎖成功,方法直接返回。
tryAcquire(arg)
final boolean nonfairTryAcquire(int acquires) { //獲取當(dāng)前線程 final Thread current = Thread.currentThread(); //獲取state變量值 int c = getState(); if (c == 0) { //沒(méi)有線程占用鎖 if (compareAndSetState(0, acquires)) { //占用鎖成功,設(shè)置獨(dú)占線程為當(dāng)前線程 setExclusiveOwnerThread(current); return true; } } else if (current == getExclusiveOwnerThread()) { //當(dāng)前線程已經(jīng)占用該鎖 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); // 更新state值為新的重入次數(shù) setState(nextc); return true; } //獲取鎖失敗 return false; }
非公平鎖tryAcquire的流程是:檢查state字段,若為0,表示鎖未被占用,那么嘗試占用,若不為0,檢查當(dāng)前鎖是否被自己占用,若被自己占用,則更新state字段,表示重入鎖的次數(shù)。如果以上兩點(diǎn)都沒(méi)有成功,則獲取鎖失敗,返回false。
2. 第二步,入隊(duì)。由于上文中提到線程A已經(jīng)占用了鎖,所以B和C執(zhí)行tryAcquire失敗,并且入等待隊(duì)列。如果線程A拿著鎖死死不放,那么B和C就會(huì)被掛起。
先看下入隊(duì)的過(guò)程。
先看addWaiter(Node.EXCLUSIVE)
/** * 將新節(jié)點(diǎn)和當(dāng)前線程關(guān)聯(lián)并且入隊(duì)列 * @param mode 獨(dú)占/共享 * @return 新節(jié)點(diǎn) */ private Node addWaiter(Node mode) { //初始化節(jié)點(diǎn),設(shè)置關(guān)聯(lián)線程和模式(獨(dú)占 or 共享) Node node = new Node(Thread.currentThread(), mode); // 獲取尾節(jié)點(diǎn)引用 Node pred = tail; // 尾節(jié)點(diǎn)不為空,說(shuō)明隊(duì)列已經(jīng)初始化過(guò) if (pred != null) { node.prev = pred; // 設(shè)置新節(jié)點(diǎn)為尾節(jié)點(diǎn) if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 尾節(jié)點(diǎn)為空,說(shuō)明隊(duì)列還未初始化,需要初始化head節(jié)點(diǎn)并入隊(duì)新節(jié)點(diǎn) enq(node); return node; }
B、C線程同時(shí)嘗試入隊(duì)列,由于隊(duì)列尚未初始化,tail==null,故至少會(huì)有一個(gè)線程會(huì)走到enq(node)。我們假設(shè)同時(shí)走到了enq(node)里。
/** * 初始化隊(duì)列并且入隊(duì)新節(jié)點(diǎn) */ private Node enq(final Node node) { //開(kāi)始自旋 for (;;) { Node t = tail; if (t == null) { // Must initialize // 如果tail為空,則新建一個(gè)head節(jié)點(diǎn),并且tail指向head if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; // tail不為空,將新節(jié)點(diǎn)入隊(duì) if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
這里體現(xiàn)了經(jīng)典的自旋+CAS組合來(lái)實(shí)現(xiàn)非阻塞的原子操作。由于compareAndSetHead的實(shí)現(xiàn)使用了unsafe類(lèi)提供的CAS操作,所以只有一個(gè)線程會(huì)創(chuàng)建head節(jié)點(diǎn)成功。假設(shè)線程B成功,之后B、C開(kāi)始第二輪循環(huán),此時(shí)tail已經(jīng)不為空,兩個(gè)線程都走到else里面。假設(shè)B線程compareAndSetTail成功,那么B就可以返回了,C由于入隊(duì)失敗還需要第三輪循環(huán)。最終所有線程都可以成功入隊(duì)。
當(dāng)B、C入等待隊(duì)列后,此時(shí)AQS隊(duì)列如下:
3. 第三步,掛起。B和C相繼執(zhí)行acquireQueued(final Node node, int arg)。這個(gè)方法讓已經(jīng)入隊(duì)的線程嘗試獲取鎖,若失敗則會(huì)被掛起。
/** * 已經(jīng)入隊(duì)的線程嘗試獲取鎖 */ final boolean acquireQueued(final Node node, int arg) { boolean failed = true; //標(biāo)記是否成功獲取鎖 try { boolean interrupted = false; //標(biāo)記線程是否被中斷過(guò) for (;;) { final Node p = node.predecessor(); //獲取前驅(qū)節(jié)點(diǎn) //如果前驅(qū)是head,即該結(jié)點(diǎn)已成老二,那么便有資格去嘗試獲取鎖 if (p == head && tryAcquire(arg)) { setHead(node); // 獲取成功,將當(dāng)前節(jié)點(diǎn)設(shè)置為head節(jié)點(diǎn) p.next = null; // 原h(huán)ead節(jié)點(diǎn)出隊(duì),在某個(gè)時(shí)間點(diǎn)被GC回收 failed = false; //獲取成功 return interrupted; //返回是否被中斷過(guò) } // 判斷獲取失敗后是否可以掛起,若可以則掛起 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) // 線程若被中斷,設(shè)置interrupted為true interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
code里的注釋已經(jīng)很清晰的說(shuō)明了acquireQueued的執(zhí)行流程。假設(shè)B和C在競(jìng)爭(zhēng)鎖的過(guò)程中A一直持有鎖,那么它們的tryAcquire操作都會(huì)失敗,因此會(huì)走到第2個(gè)if語(yǔ)句中。我們?cè)倏聪聅houldParkAfterFailedAcquire和parkAndCheckInterrupt都做了哪些事吧。
/** * 判斷當(dāng)前線程獲取鎖失敗之后是否需要掛起. */ private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { //前驅(qū)節(jié)點(diǎn)的狀態(tài) int ws = pred.waitStatus; if (ws == Node.SIGNAL) // 前驅(qū)節(jié)點(diǎn)狀態(tài)為signal,返回true return true; // 前驅(qū)節(jié)點(diǎn)狀態(tài)為CANCELLED if (ws > 0) { // 從隊(duì)尾向前尋找第一個(gè)狀態(tài)不為CANCELLED的節(jié)點(diǎn) do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { // 將前驅(qū)節(jié)點(diǎn)的狀態(tài)設(shè)置為SIGNAL compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } /** * 掛起當(dāng)前線程,返回線程中斷狀態(tài)并重置 */ private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); }
線程入隊(duì)后能夠掛起的前提是,它的前驅(qū)節(jié)點(diǎn)的狀態(tài)為SIGNAL,它的含義是“Hi,前面的兄弟,如果你獲取鎖并且出隊(duì)后,記得把我喚醒!”。所以shouldParkAfterFailedAcquire會(huì)先判斷當(dāng)前節(jié)點(diǎn)的前驅(qū)是否狀態(tài)符合要求,若符合則返回true,然后調(diào)用parkAndCheckInterrupt,將自己掛起。如果不符合,再看前驅(qū)節(jié)點(diǎn)是否>0(CANCELLED),若是那么向前遍歷直到找到第一個(gè)符合要求的前驅(qū),若不是則將前驅(qū)節(jié)點(diǎn)的狀態(tài)設(shè)置為SIGNAL。
整個(gè)流程中,如果前驅(qū)結(jié)點(diǎn)的狀態(tài)不是SIGNAL,那么自己就不能安心掛起,需要去找個(gè)安心的掛起點(diǎn),同時(shí)可以再?lài)L試下看有沒(méi)有機(jī)會(huì)去嘗試競(jìng)爭(zhēng)鎖。
最終隊(duì)列可能會(huì)如下圖所示
線程B和C都已經(jīng)入隊(duì),并且都被掛起。當(dāng)線程A釋放鎖的時(shí)候,就會(huì)去喚醒線程B去獲取鎖啦。
3.3.2 unlock()
unlock相對(duì)于lock就簡(jiǎn)單很多。源碼如下
public void unlock() { sync.release(1); } public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
如果理解了加鎖的過(guò)程,那么解鎖看起來(lái)就容易多了。流程大致為先嘗試釋放鎖,若釋放成功,那么查看頭結(jié)點(diǎn)的狀態(tài)是否為SIGNAL,如果是則喚醒頭結(jié)點(diǎn)的下個(gè)節(jié)點(diǎn)關(guān)聯(lián)的線程,如果釋放失敗那么返回false表示解鎖失敗。這里我們也發(fā)現(xiàn)了,每次都只喚起頭結(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)關(guān)聯(lián)的線程。
最后我們?cè)倏聪聇ryRelease的執(zhí)行過(guò)程
/** * 釋放當(dāng)前線程占用的鎖 * @param releases * @return 是否釋放成功 */ protected final boolean tryRelease(int releases) { // 計(jì)算釋放后state值 int c = getState() - releases; // 如果不是當(dāng)前線程占用鎖,那么拋出異常 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { // 鎖被重入次數(shù)為0,表示釋放成功 free = true; // 清空獨(dú)占線程 setExclusiveOwnerThread(null); } // 更新state值 setState(c); return free; }
這里入?yún)?。tryRelease的過(guò)程為:當(dāng)前釋放鎖的線程若不持有鎖,則拋出異常。若持有鎖,計(jì)算釋放后的state值是否為0,若為0表示鎖已經(jīng)被成功釋放,并且則清空獨(dú)占線程,最后更新state值,返回free。
3.3.3 小結(jié)
用一張流程圖總結(jié)一下非公平鎖的獲取鎖的過(guò)程。
3.4 FairSync
公平鎖和非公平鎖不同之處在于,公平鎖在獲取鎖的時(shí)候,不會(huì)先去檢查state狀態(tài),而是直接執(zhí)行aqcuire(1),這里不再贅述。
4 超時(shí)機(jī)制
在ReetrantLock的tryLock(long timeout, TimeUnit unit) 提供了超時(shí)獲取鎖的功能。它的語(yǔ)義是在指定的時(shí)間內(nèi)如果獲取到鎖就返回true,獲取不到則返回false。這種機(jī)制避免了線程無(wú)限期的等待鎖釋放。那么超時(shí)的功能是怎么實(shí)現(xiàn)的呢?我們還是用非公平鎖為例來(lái)一探究竟。
public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1, unit.toNanos(timeout)); }
還是調(diào)用了內(nèi)部類(lèi)里面的方法。我們繼續(xù)向前探究
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout); }
這里的語(yǔ)義是:如果線程被中斷了,那么直接拋出InterruptedException。如果未中斷,先嘗試獲取鎖,獲取成功就直接返回,獲取失敗則進(jìn)入doAcquireNanos。tryAcquire我們已經(jīng)看過(guò),這里重點(diǎn)看一下doAcquireNanos做了什么。
/** * 在有限的時(shí)間內(nèi)去競(jìng)爭(zhēng)鎖 * @return 是否獲取成功 */ private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { // 起始時(shí)間 long lastTime = System.nanoTime(); // 線程入隊(duì) final Node node = addWaiter(Node.EXCLUSIVE); boolean failed = true; try { // 又是自旋! for (;;) { // 獲取前驅(qū)節(jié)點(diǎn) final Node p = node.predecessor(); // 如果前驅(qū)是頭節(jié)點(diǎn)并且占用鎖成功,則將當(dāng)前節(jié)點(diǎn)變成頭結(jié)點(diǎn) if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return true; } // 如果已經(jīng)超時(shí),返回false if (nanosTimeout <= 0) return false; // 超時(shí)時(shí)間未到,且需要掛起 if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) // 阻塞當(dāng)前線程直到超時(shí)時(shí)間到期 LockSupport.parkNanos(this, nanosTimeout); long now = System.nanoTime(); // 更新nanosTimeout nanosTimeout -= now - lastTime; lastTime = now; if (Thread.interrupted()) //相應(yīng)中斷 throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
doAcquireNanos的流程簡(jiǎn)述為:線程先入等待隊(duì)列,然后開(kāi)始自旋,嘗試獲取鎖,獲取成功就返回,失敗則在隊(duì)列里找一個(gè)安全點(diǎn)把自己掛起直到超時(shí)時(shí)間過(guò)期。這里為什么還需要循環(huán)呢?因?yàn)楫?dāng)前線程節(jié)點(diǎn)的前驅(qū)狀態(tài)可能不是SIGNAL,那么在當(dāng)前這一輪循環(huán)中線程不會(huì)被掛起,然后更新超時(shí)時(shí)間,開(kāi)始新一輪的嘗試
5 總結(jié)
ReentrantLock的核心功能講解差不多落下帷幕,理解AQS,就很容易理解ReentrantLock的實(shí)現(xiàn)原理。文中慘雜著筆者的個(gè)人理解,如有不正之處,還望指正。
看完上述內(nèi)容,你們掌握深入淺析ReentrantL中實(shí)現(xiàn)ock的原理的方法了嗎?如果還想學(xué)到更多技能或想了解更多相關(guān)內(nèi)容,歡迎關(guān)注億速云行業(yè)資訊頻道,感謝各位的閱讀!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。