ReentrantLock
ReentrantLock是一種可重入的互斥鎖，它的行為和作用與關鍵字synchronized有些類似，在并發場景下可以讓多個線程按照一定的順序訪問同一資源。相比synchronized，ReentrantLock多了可擴展的能力，比如我們可以創建一個名為MyReentrantLock的類繼承ReentrantLock，并重寫部分方法使其更加高效。
當一個線程調用ReentrantLock.lock()方法時，如果ReentrantLock沒有被其他線程持有，且不存在額外的線程與當前線程競爭ReentrantLock，調用ReentrantLock.lock()方法后當前線程會占有此鎖并立即返回，ReentrantLock內部會維護當前線程對鎖的引用計數，當線程獲取鎖時會增加其線程對鎖的引用計數，當線程釋放鎖時會減少線程對鎖的引用計數，當前線程如果在占有鎖之后，又重復獲取鎖，則會增加鎖的引用計數，當鎖的引用計數為0的時候，代表當前線程完全釋放鎖。需要注意的是，只有占有鎖的線程才會增加鎖的引用計數，當鎖被占據時，如果有其他線程要競爭鎖，ReentrantLock會把其他線程加入一個競爭鎖的隊列，并讓線程陷入阻塞，直到占據鎖的線程釋放了鎖，ReentrantLock才會喚醒隊列中的線程重新競爭鎖。
我們用下面的例子來加深對于鎖的理解，假設我們的進程內目前沒有任何線程競爭lock，此時鎖的引用計數為0，有一個線程Thread-1調用完下面<1>處的lock()方法成功占有鎖，此時鎖的引用計數由0變為1。之后Thread-1調用了<2>處的methodB()方法，methodB()的<4>處又獲取了一次鎖，由于lock已經被Thread-1占據，所以這里簡單的對鎖的引用計數+1即可，此時鎖的引用計數為2，Thread-1執行完methodB()的方法體后，執行<5>處的unlock()方法釋放鎖，這里對鎖的引用計數-1，由2變為1。在調用完methodB后，執行methodA的方法體，最后執行<3>處的unlock()方法，將鎖的引用計數由1變為0，Thread-1完全釋放鎖。此時，鎖變為無主狀態。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
private final ReentrantLock lock = new ReentrantLock(); public void methodA() { try { lock.lock();//<1> methodB();//<2> //methodA body... } finally { lock.unlock();//<3> } } public void methodB() { try { lock.lock();//<4> //methodB body... } finally { lock.unlock();//<5> }} |
ReentrantLock提供了isHeldByCurrentThread()和getHoldCount()兩個方法，前者用于判斷鎖是否被當先調用線程持有，如果被當前調用線程持有則返回true；后者不僅會判斷鎖是否被當前線程持有，還會返回鎖相對于當前線程的引用計數，畢竟鎖是可重入的，如果鎖沒有被任何線程持有，或者被不是持有鎖的線程調用getHoldCount()方法，就會返回0。
這兩個方法的實現原理也很簡單，我們知道在Java中可以調用Thread.currentThread()來獲取當前線程對象。當我們調用ReentrantLock.lock()方法成功獲取鎖之后，ReentrantLock內部會用一個獨占線程（exclusiveOwnerThread）字段來標識當前占用鎖的Thread線程對象，如果線程釋放了鎖且鎖的引用計數為0，則將獨占線程字段標記為null。當要判斷鎖是否被當前線程持有，或者鎖相對于當前線程的引用計數，則獲取調用方線程的Thread對象，和內部的獨占線程字段做下對比，如果兩者的引用相等，代表當前線程占用了鎖，如果引用不相等，則表示當前所可能處于無主狀態，或者鎖被其他線程持有。
如下面的代碼，我們希望只有持有lock的線程才可以執行methodB()和methodC()方法，就可以用isHeldByCurrentThread()和getHoldCount()進行判斷。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
private final ReentrantLock lock = new ReentrantLock(); public void methodA() { try { lock.lock(); methodB(); methodC(); //methodA body... } finally { lock.unlock(); }} public void methodB() { if (lock.getHoldCount() != 0) { //methodB body... }} public void methodC() { if (lock.isHeldByCurrentThread()) { //methodC body... }} |
需要注意的一點是，官方有給出isHeldByCurrentThread()和getHoldCount()兩個方法的使用范圍，僅針對于debug和測試。真正的生產環境如果有重入鎖的需要，官方還是推薦用try{}finally{}這一套，在try代碼塊里獲取鎖，在finally塊中釋放鎖。
創建ReentrantLock對象時，如果使用的是無參構造方法，則默認創建非公平鎖（NonfairSync），如果調用的是ReentrantLock(boolean fair)有參構造方法，fair為true則創建公平鎖（FairSync）。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
public class ReentrantLock implements Lock, java.io.Serializable { //... //默認創建非公平鎖 public ReentrantLock() { sync = new NonfairSync(); } //根據參數指定創建公平鎖或非公平鎖，true為公平鎖。 public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); } //...} |
之前說過，當有多個線程競爭鎖時，獲取鎖失敗的線程，會形成一個隊列。如果有多個線程競爭公平鎖時，會優先把鎖分配給等待鎖時間最長的線程，即隊頭的線程，隊列中越往后的線程等待鎖的時間越短，排在隊尾的線程等待時間最短。如果使用的是非公平鎖，則不保證會按照等待時長順序將鎖分配。在多線程的場景下，公平鎖在吞吐量方面的表現不如非公平鎖，但兩者在獲得鎖和保證不饑餓的差異并不大。
需要注意的是，公平鎖不能保證線程調度的公平性，競爭公平鎖的多個線程中，可能會出現一個線程連續多次獲得鎖的情況。比如：Thread-1、Thread-2都要競爭同一個鎖（lock），但此時鎖已經被其他線程占據，Thread-1、Thread-2競爭失敗，預備進入等待隊列，這時Thread-1、Thread-2的CPU時間片消耗完畢被掛起，而其他線程剛好釋放鎖將鎖變為無主狀態，此時Thread-3搶鎖成功，并調用下面的doThread3()方法，連續10次獲取鎖并釋放鎖將鎖變為無主狀態。這種情況，就是上面說的公平鎖無法保證線程調度的公平性，按照順序，Thread-3在Thread-1、Thread-2競爭失敗后才開始競爭，按理鎖的分配順序應該是Thread-1->Thread-2->Thread-3，但由于線程的調度問題，Thread-1、Thread-2尚未入隊，而鎖被釋放后剛好被Thread-3“撿漏”
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
public void methodA() { try { lock.lock(); //methodA body... } finally { lock.unlock(); }} public void doThread3() { for (int i = 0; i < 10; i++) { methodA(); }} |
除了調用ReentrantLock.lock()以阻塞的方式直到獲取鎖，ReentrantLock還提供了tryLock()和tryLock(long timeout, TimeUnit unit)兩個方法來搶鎖。我們看下面的代碼，相信很多同學看到這兩個方法后也能知道這兩個方法和lock()方法的區別，tryLock()會嘗試競爭鎖，如果鎖已被其他線程占用，則競爭失敗，返回false，如果競爭成功，則返回true。tryLock(long timeout, TimeUnit unit)如果競爭鎖失敗后，會先進入等待隊列，如果在過期前能競爭到鎖，則返回true，如果在過期時間內都無法搶到鎖，則返回false。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public void methodD() { boolean hasLock = false; try { hasLock = lock.tryLock();//<1>非計時 if (!hasLock) {//沒有搶到鎖則退出 return; } //methodD body... } finally { if (hasLock) { lock.unlock(); } }} public void methodE() { boolean hasLock = false; try { hasLock = lock.tryLock(5, TimeUnit.SECONDS);//<2>計時 if (!hasLock) {//沒有搶到鎖則退出 return; } //methodE body... } catch (InterruptedException e) { e.printStackTrace(); } finally { if (hasLock) { lock.unlock(); } }} |
需要注意的是：不管是公平鎖還是非公平鎖，不計時tryLock()都不能保證公平性，如果鎖可用，即時其他線程正在等待鎖，也會搶鎖成功。
ReentrantLock內部會用一個int字段來標識鎖的引用次數，因此，ReentrantLock雖然作為可重入鎖，但它的最大可重入次數為2147483647（即：MaxInt32，2^31-1），不管我們是以遞歸或者是循環亦或者其他方式，一旦我們重復獲取鎖的次數超過這個次數，ReentrantLock就會拋出異常。
至此，我們了解了ReentrantLock的簡單應用。下面，就請大家一起跟隨筆者了解ReentrantLock的實現原理。下面的代碼是筆者從ReentrantLock節選的部分代碼，可以看到先前我們調用加鎖（lock、lockInterruptibly、tryLock）、解鎖（unlock）的代碼，最后都會調用sync對象的方法，sync對象的類型是一個抽象類，在我們創建ReentrantLock對象時，會根據構造函數決定sync是公平鎖（FairSync），還是非公平鎖（NonfairSync），FairSync和NonfairSync都繼承自Sync，所以ReentrantLock在創建好具體的Sync對象后，便不再管關心公平鎖的邏輯或者是非公平鎖的邏輯，ReentrantLock只知道抽象類Sync實現了它所需要的功能，這個功能是公平亦或是非公平，由具體的實現子類來關心。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public void methodD() { boolean hasLock = false; try { hasLock = lock.tryLock();//<1>非計時 if (!hasLock) {//沒有搶到鎖則退出 return; } //methodD body... } finally { if (hasLock) { lock.unlock(); } }} public void methodE() { boolean hasLock = false; try { hasLock = lock.tryLock(5, TimeUnit.SECONDS);//<2>計時 if (!hasLock) {//沒有搶到鎖則退出 return; } //methodE body... } catch (InterruptedException e) { e.printStackTrace(); } finally { if (hasLock) { lock.unlock(); } }} |
鑒于ReentrantLock的無參構造函數是創建一個非公平鎖，可見官方更傾向于我們使用非公平鎖，這里，我們就先從非公平鎖開始介紹。
當ReentrantLock為非公平鎖時，調用lock()方法會直接調用sync.acquire(1)，NonfairSync和Sync兩個類都沒有實現acquire(int arg)，這個方法是由AbstractQueuedSynchronizer（抽象隊列同步器，下面簡稱：AQS）實現的，也就是Sync的父類。
當線程競爭鎖時，會先調用tryAcquire(arg)方法試圖占有鎖，AQS將tryAcquire(int arg)的實現交由子類，由子類決定是以公平還是非公平的方式占有鎖，如果競爭成功tryAcquire(arg)則返回true，!tryAcquire(arg)的結果為false，于是就不會再調用<1>處后續的判斷，直接返回。如果占有鎖失敗，這里會先調用addWaiter(Node mode)方法，將當前調用線程封裝成一個Node對象，再調用acquireQueued(final Node node, int arg)將Node對象加入到等待隊列中，并使線程陷入阻塞。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
//java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//<1> selfInterrupt(); } //AbstractQueuedSynchronizer將tryAcquire(int arg)的實現交由子類//java.util.concurrent.locks.AbstractQueuedSynchronizer#tryAcquire protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); } |
我們先來看NonfairSync實現的tryAcquire(int acquires)方法，這里NonfairSync也是調用其父類Sync的nonfairTryAcquire(int acquires)方法。在AQS內部會維護一個volatile int state，可重入互斥鎖會用這個字段存儲占有鎖的線程對鎖的引用計數，即重復獲取鎖的次數。如果state為0，代表鎖目前沒有被任何線程占有，這里會用CAS的方式設置鎖的引用計數，如果設置成功，則執行<2>處的代碼將獨占線程（exclusiveOwnerThread）的引用指向當前調用線程，然后返回true表示加鎖成功。
如果當前state不為0，代表有線程正獨占此鎖，會在<3>處判斷當前線程是否是獨占線程，如果是的話則在<4>處增加鎖的引用計數，這里同樣是修改state的值，但不需要像<1>處那樣用CAS的方式，因為<4>處的代碼只有獨占線程才可以執行，其他線程都無法執行。需要注意的一點是，state為int類型，最大值為：2^31-1，如果超過這個值state就會變為負數，就會報錯。如果一個線程在競爭鎖的時候，發現state不為0，且當前線程不是獨占線程，則會返回false，表示搶鎖失敗。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
//當調用AQS的acquire(int arg)時，會先調用由子類實現的tryAcquire(int acquires)方法//java.util.concurrent.locks.ReentrantLock.NonfairSync#tryAcquire protected final boolean tryAcquire(int acquires) { //這里會調用父類Sync的nonfairTryAcquire(int acquires)方法 return nonfairTryAcquire(acquires); }//java.util.concurrent.locks.ReentrantLock.Sync#nonfairTryAcquire final boolean nonfairTryAcquire(int acquires) { //獲取當前線程對象 final Thread current = Thread.currentThread(); //這里會獲取父類AQS的state字段，在可重入互斥鎖里，state表示占有鎖的線程的引用計數 int c = getState(); //如果state為0，表示目前鎖是無主狀態 if (c == 0) { //如果鎖處于無主狀態，則用CAS修改state，如果修改成功，表示占有鎖成功 if (compareAndSetState(0, acquires)) {//<1> //占有鎖成功后，這里會設置鎖的獨占線程 setExclusiveOwnerThread(current);//<2> return true; } } else if (current == getExclusiveOwnerThread()) {//<3>如果state不為0，代表現在有線程占據鎖，如果請求鎖的線程和獨占線程是同一個線程，則增加當前線程對鎖的引用計數 //鎖的最大可重入次數為(2^31-1)，超過這個最大范圍，int就會變為負數，判斷nextc為負數時報錯。 int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); //重新設置state的值 setState(nextc);//<4> return true; } return false; } public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { //... //在可重入互斥鎖中，state代表獨占線程當前的重入次數 private volatile int state; protected final int getState() { return state; } protected final void setState(int newState) { state = newState; } //...} public abstract class AbstractOwnableSynchronizer implements java.io.Serializable { //... //獨占線程，當有線程占據可重入互斥鎖時，會用此字段存儲占有鎖的線程 private transient Thread exclusiveOwnerThread; protected final void setExclusiveOwnerThread(Thread thread) { exclusiveOwnerThread = thread; } protected final Thread getExclusiveOwnerThread() { return exclusiveOwnerThread; }} |
按照AbstractQueuedSynchronizer.acquire(int arg)的邏輯，如果搶鎖失敗，會繼而執行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)這段代碼。這里我們需要先來了解下Node的數據結構，Node類是AQS的一個靜態內部類。如果眼尖的同學看到下面的prev和next，一定能很快猜出這就是我們先前所說的等待隊列，等待隊列實質上是一個雙端鏈表，即每個節點都可以知道自己的前驅，也可以知道自己的后繼。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
//java.util.concurrent.locks.AbstractQueuedSynchronizer.Node static final class Node { static final Node SHARED = new Node(); static final Node EXCLUSIVE = null; static final int CANCELLED = 1; static final int SIGNAL = -1; //... volatile int waitStatus; volatile Node prev; volatile Node next; volatile Thread thread; Node nextWaiter; //... //返回當前節點的前驅節點 final Node predecessor() { Node p = prev; if (p == null) throw new NullPointerException(); else return p; } Node() {} //... //創建Node節點 Node(Node nextWaiter) {//<1> this.nextWaiter = nextWaiter; THREAD.set(this, Thread.currentThread()); } } |
這里簡單介紹下Node的字段：
- prev指向當前節點的前驅節點，next指向當前節點的后繼節點。
- thread字段在調用<1>處的構造方法時，會將thread指向當前調用線程的Thread對象。
- waitStatus（等待狀態）初始值為0，當waitStatus為SIGNAL（-1）時，表示當前節點的后繼節點所指向的線程（node.next.thread）陷入阻塞，當前節點如果被移除（CANCELLED）或在占有鎖后要釋放鎖的時候，需要喚醒后繼節點的線程。這里有多種可能導致當前節點的等待狀態變為移除，比如調用tryLock(long timeout, TimeUnit unit) 超時會獲取到鎖，或者調用lockInterruptibly()后線程被中斷。
- nextWaiter可以用來表示一個節點的線程到底是獨占線程（EXCLUSIVE）還是共享線程（SHARED），獨占線程一般用于可重入互斥鎖（ReentrantLock）或者可重入讀寫鎖（ReentrantReadWriteLock）的寫鎖，而共享線程則表示當前線程是可以和其他共享線程一起共享資源的，一般用于可重入讀寫鎖的讀鎖。
如果對上面Node字段還有不理解的地方不用心急，筆者在后面還會和大家一起深入了解這幾個字段。
在簡單了解了Node的數據結構后，我們來看看AQS是如何將一個線程封裝成一個Node對象，并將其加入到等待隊列。addWaiter(Node mode)會根據傳入的參數node，決定創建的節點是獨占節點還是共享節點，先前ReentrantLock傳入的是Node.EXCLUSIVE，所以這里是獨占節點，在執行完<1>處的代碼后，節點創建完畢，節點的thread字段也保存了當前線程對象的引用。之后會進入<2>處的循環，這里是通過CAS自旋的方式將節點加入到等待隊列，之所以用這種方式是因為可能存在多個線程同時要入隊的情況，用CAS自旋保證每個節點的前驅和后繼的有序性。當節點要入隊時，會先獲取尾節點，如果在<3>處判斷尾節點不為null，則將當前節點的前驅指向尾節點，并用CAS的方式設置當前節點為設置為尾節點，如果原先的尾節點（oldTail）的指向沒有被任何線程修改，這里用CAS將當前節點設置成尾節點就會成功，于是原先尾節點的后繼指向當前節點，當前節點入隊成功。但我們也要考慮尾節點為null的情況，即第一個進入等待隊列的節點，此時頭節點（header）和尾節點（tail）都為null，這里就會執行<4>處的分支，進行隊列初始化。初始化隊列的時候，同樣存在并發問題，所以這里依舊用CAS初始化頭節點成功，再將頭節點指向的Node對象賦值給尾節點。初始化隊列完畢后，會再開始新的一輪循環，用CAS的方式嘗試將節點入隊，入隊成功后，則返回當前節點。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { //... private transient volatile Node head;//等待隊列的頭節點 private transient volatile Node tail;//等待隊列的尾節點 //... private Node addWaiter(Node mode) { //為競爭鎖的線程創建一個Node對象，并用Node.thread字段存儲調用線程Thread對象 Node node = new Node(mode);//<1> for (;;) {//<2> Node oldTail = tail; if (oldTail != null) {//<3> node.setPrevRelaxed(oldTail); if (compareAndSetTail(oldTail, node)) { oldTail.next = node; return node; } } else {//<4> initializeSyncQueue(); } } } private final void initializeSyncQueue() { Node h; if (HEAD.compareAndSet(this, null, (h = new Node()))) tail = h; } private final boolean compareAndSetTail(Node expect, Node update) { return TAIL.compareAndSet(this, expect, update); } //...} |
在執行完addWaiter(Node.EXCLUSIVE)確定節點入隊后，就要將返回節點傳入到方法：acquireQueued(final Node node, int arg)。之前我們說過，搶鎖失敗的節點會進入一個等待隊列，等待鎖的分配，我們已經在addWaiter(Node mode)看到線程是如何入隊的，那接下來就要看看線程是如何等待鎖的分配。在看acquireQueued(final Node node, int arg)之前，我們先來思考下如果是我們自己會如何設計將鎖分配給線程？最簡單的做法是每個線程都在一個死循環中去輪詢鎖的狀態，如果發現鎖處于無主狀態并搶鎖成功，線程則跳出循環訪問資源。但這個做法有個缺點就是會消耗CPU時間片，尤其對于一些優先級不高的線程，相比于優先級高的線程它們可能永遠無法競爭到鎖，永遠訪問不到資源處于饑餓狀態。那么有沒有相比死循環更好的做法呢？我們是否可以先把一個入隊的線程阻塞起來，先讓它不要消耗寶貴的CPU時間片，當占據鎖的線程完全釋放鎖（state變為0）時，則去喚醒隊列中等待時長最長的線程，這樣也不用擔心優先級低的線程無法與優先級高的線程競爭鎖，導致處于饑餓狀態，一舉兩得。
這里我們還要再加深下對等待隊列Node的理解才能往下看acquireQueued(final Node node, int arg)，大家思考下，Node中的thread字段是用來指向競爭鎖的線程對象，通過這個對象，我們可以用釋放鎖的線程喚醒等待鎖的線程，占用鎖的線程在完全釋放鎖將鎖變為無主狀態后，喚醒等待鎖的線程，這個等待鎖的線程如果成功占據了鎖，是否可以將本身線程中Node.thread置為null？此刻線程已經占據了鎖，它不會再陷入阻塞，也不需要有其他的線程來喚醒自身。所以等待隊列的頭節點的thread(header.thread)字段永遠為null，因為鎖被頭節點的線程所占用。
當然，也可能出現鎖被占用但頭節點(header)本身就為null，這種情況一般出現在我們初始化好一個ReentrantLock后，只有一個線程占有了鎖，此時調用tryAcquire(int acquires)會調用ReentrantLock.Sync.nonfairTryAcquire(int acquires)方法，這個方法只會簡單修改state狀態，并不會新增一個頭節點。除非鎖已有線程占據，且出現新的線程競爭鎖，這時候新的線程在進入等待隊列的時候，會初始化隊列，為本身占據鎖的線程補上一個頭節點，初始化隊列的時候調用的是Node的無參構造方法，所以頭節點的thread字段為null，表示鎖被當前頭節點原先指向的線程所占據。
在了解這些基本知識后，下面我們終于可以來看看大家迫不及待的acquireQueued(final Node node, int arg)了。當把封裝了當前線程的Node對象傳入到acquireQueued(final Node node, int arg)方法時，并不會立即阻塞當前線程等待其他線程喚醒。這里會先在<1>處獲取當前節點的前驅節點p，判斷p是不是頭節點，如果p是頭節點，則當前線程即有占有鎖的可能。因為占據鎖的線程會先釋放鎖，再通知隊列中的線程搶鎖。所以會存在當前節點入隊前鎖已被釋放的情況，于是判斷前驅節點p是頭節點，會再調用tryAcquire(int acquires)方法搶鎖，如果搶鎖成功，就可以按照我們上面所說的套路，調用setHead(Node node)將當前節點設置為頭節點，設置當前節點的線程引用為null，然后返回。
如果當前節點的前驅節點不是頭節點，這里就要調用shouldParkAfterFailedAcquire(Node pred, Node node)設置前驅節點的等待狀態（waitStatus），先前說過，這個等待狀態可以用來表示下個節點的阻塞狀態。假設有一個鎖已經被其他線程占有，Thread-1、Thread-2要來搶鎖，此時必然是搶鎖失敗的，這里會把Thread-1、Thread-2分別封裝成Node1和Node2并進行入隊，Node1和Node2初始的等待狀態都為0，假定Node1先Node2入隊，Node1為Node2的前驅節點（即：Node2.prev=Node1），Node1不是頭節點，所以不會去搶鎖，這里直接進入<2>處分支的shouldParkAfterFailedAcquire(Node pred, Node node)方法，Node1的初始等待狀態為0，所以<3>處和<5>處的分支是進不去的，只能進入<4>處的分支，將Node1的等待狀態設置為SIGNAL，表示Node1的后繼節點處于等待喚醒狀態，然后返回false，于是<2>處的判斷不成立，又開始新的一輪循環，假定頭節點的線程依舊沒釋放鎖，Node1依舊不是頭節點，還是直接執行shouldParkAfterFailedAcquire(Node pred, Node node)方法，此時判斷Node2的前驅節點Node1的等待狀態為-1，表示可以阻塞Node1后繼節點Node2所指向的線程，所以這里會返回true，進入<2>處的分支，調用parkAndCheckInterrupt()方法，在這個方法中會調用LockSupport.park(Object blocker)阻塞當前的調用線程，直到有其他線程調用LockSupport.unpark(Node2.thread)喚醒Node2被阻塞的線程，或Node2.thread被中斷才會退出parkAndCheckInterrupt()。我們注意到在<5>處有一個判斷，前驅節點的等待狀態>0，一般狀態為CANCELLED（1），表示前驅節點被移除。之所以會存在被移除的節點，是因為我們可能以tryLock(long timeout, TimeUnit unit)的方式往等待隊列中添加節點，如果超時還未獲得鎖，這個節點就要被移除；我們還可能用lockInterruptibly()的方式往等待隊列中添加節點，如果節點所對應的線程被中斷，這個節點也處于被移除狀態。所以<5>處如果發現前驅節點的等待狀態大于0，會一直往前驅節點遍歷直到找到等待狀態<=0的節點將其作為前驅節點，并將前驅節點的后繼指向當前節點。要注意的是，等待狀態為-1時，代表當前節點的后繼節點等待喚醒，>0的時候，代表當前節點被移除，前者的狀態與后繼節點有關，后者的狀態僅與自身有關。如果在自旋期間線程出現其他異常，則會調用<6>處的代碼將節點從等待隊列移除，并拋出異常。cancelAcquire(Node node)會在后面介紹，這里我們只要先知道這是一個將節點從隊列中移除的方法。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { //... private transient volatile Node head; //... final boolean acquireQueued(final Node node, int arg) { boolean interrupted = false; try { for (;;) { final Node p = node.predecessor();//<1> if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC return interrupted; } if (shouldParkAfterFailedAcquire(p, node))//<2> interrupted |= parkAndCheckInterrupt(); } } catch (Throwable t) { cancelAcquire(node);//<6> if (interrupted) selfInterrupt(); throw t; } } //... //設置當前節點為頭節點，此時可以清空頭節點指向的線程引用 private void setHead(Node node) { head = node; node.thread = null; node.prev = null; } //... private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL)//<3> /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; if (ws > 0) {//<5> /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else {//<4> /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ pred.compareAndSetWaitStatus(ws, Node.SIGNAL); } return false; } //... private final boolean parkAndCheckInterrupt() { //阻塞調用線程，可調用LockSupport.unpark(Thread thread)喚醒或由線程中斷喚醒。 LockSupport.park(this); //返回線程是否由中斷喚醒，返回true為被中斷喚醒，但此方法會清除線程的中斷標記 return Thread.interrupted(); } //...} |
能從boolean acquireQueued(final Node node, int arg)方法中返回的線程，都是成功占有鎖的線程，但返回結果分當前線程是否被中斷，true為被中斷。可能存在這樣一種情況，前一個線程釋放鎖完畢后，即將喚醒后一個線程，此時后一個線程被中斷喚醒，后一個線程發現其Node節點的前驅節點為頭節點，且鎖為無主狀態，于是搶鎖成功直接返回。這里要標記線程的中斷狀態interrupted，因為線程會從parkAndCheckInterrupt()中被喚醒，最后會執行Thread.interrupted()返回當前線程是否由中斷喚醒，但Thread.interrupted()會清除中斷標記，所以在占據鎖之后會根據返回的interrupted狀態，決定是否設置線程的中斷狀態。如果一個線程在調用acquireQueued(final Node node, int arg)方法的后都未被中斷，直到前一個線程調用LockSupport.unpark(Thread thread)喚醒該線程，那么這個線程就不是用中斷的形式喚醒，也就不用設置線程的中斷狀態。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { //... public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //根據acquireQueued()的返回，決定是否設置線程的中斷標記 selfInterrupt(); } //... static void selfInterrupt() { Thread.currentThread().interrupt(); } //...} |
到此這篇關于Java源碼解析之詳解ReentrantLock的文章就介紹到這了,更多相關ReentrantLock源碼解析內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://www.cnblogs.com/beiluowuzheng/p/14948225.html
