ArrayList源碼分析--jdk1.8
LinkedList源碼分析--jdk1.8
HashMap源碼分析--jdk1.8
AQS源碼分析--jdk1.8
ReentrantLock源碼分析--jdk1.8
成都創(chuàng)新互聯(lián)成都網(wǎng)站建設(shè)按需求定制設(shè)計(jì)，是成都網(wǎng)站營(yíng)銷公司,為木托盤提供網(wǎng)站建設(shè)服務(wù),有成熟的網(wǎng)站定制合作流程，提供網(wǎng)站定制設(shè)計(jì)服務(wù)：原型圖制作、網(wǎng)站創(chuàng)意設(shè)計(jì)、前端HTML5制作、后臺(tái)程序開(kāi)發(fā)等。成都網(wǎng)站制作熱線：13518219792
??1. AQS是一個(gè)基于FIFO隊(duì)列，可以用于構(gòu)建鎖或者其他相關(guān)同步裝置的基礎(chǔ)框架。
?2. AQS提供了雙向鏈表。
?3. AQS分為共享模式和獨(dú)占模式。
?4.AQS基于volatile內(nèi)存可見(jiàn)性和CAS原子性操作實(shí)現(xiàn)線程間通信操作。
??數(shù)據(jù)結(jié)構(gòu)是集合的精華所在，數(shù)據(jù)結(jié)構(gòu)往往也限制了集合的作用和側(cè)重點(diǎn)，了解各種數(shù)據(jù)結(jié)構(gòu)是我們分析源碼的必經(jīng)之路。
?AQS的數(shù)據(jù)結(jié)構(gòu)如下：雙向鏈表
?
?AQS實(shí)現(xiàn)共享資源的訪問(wèn)控制基礎(chǔ)：
? ??1.state字段，即同步器狀態(tài)字段。用于共享資源的訪問(wèn)控制
? ??2.CLH隊(duì)列，F(xiàn)IFO等待隊(duì)列，存放競(jìng)爭(zhēng)失敗的線程。通常CLH隊(duì)列是一個(gè)自旋隊(duì)列，AQS以阻塞的方式實(shí)現(xiàn)
? ??CLH隊(duì)列的使用：
自旋鎖
學(xué)習(xí)了解自旋鎖之前先回顧一下互斥鎖
互斥鎖
線程在獲取互斥鎖的時(shí)候，如果發(fā)現(xiàn)鎖已經(jīng)被其它線程占有，那么線程就會(huì)驚醒休眠，然后在適當(dāng)?shù)臅r(shí)機(jī)（比如喚醒）在獲取鎖。
自旋鎖
那么自旋鎖顧名思義就是“自旋”。就是當(dāng)一個(gè)線程在嘗試獲取鎖失敗之后，線程不會(huì)休眠或者掛起，而是一直在循環(huán)檢測(cè)鎖是否被其它線程釋放。
區(qū)別
互斥鎖就是開(kāi)始開(kāi)銷要大于自旋鎖。臨界區(qū)持鎖時(shí)間的大小并不會(huì)對(duì)互斥鎖的開(kāi)銷造成影響，而自旋鎖是死循環(huán)檢測(cè)，加鎖全程消耗cpu，起始開(kāi)銷雖然低于互斥鎖，但是隨著持鎖時(shí)間，加鎖的開(kāi)銷是線性增長(zhǎng)。
適用的情況
互斥鎖用于臨界區(qū)持鎖時(shí)間比較長(zhǎng)的操作，比如下面這些情況都可以考慮臨界區(qū)有IO操作
臨界區(qū)代碼復(fù)雜或者循環(huán)量大
臨界區(qū)競(jìng)爭(zhēng)非常激烈
單核處理器
自旋鎖就主要用在臨界區(qū)持鎖時(shí)間非常短且CPU資源不緊張的情況下。當(dāng)遞歸調(diào)用時(shí)有可能造成死鎖。
線程（節(jié)點(diǎn)）隊(duì)列
了解了自旋鎖之后，在學(xué)習(xí)ReentrantLock的時(shí)候，一個(gè)線程在等待鎖的時(shí)候會(huì)被封裝成一個(gè)Node節(jié)點(diǎn)，然后加入一個(gè)隊(duì)列中并檢測(cè)前一個(gè)節(jié)點(diǎn)是否是頭節(jié)點(diǎn)，并且嘗試獲取鎖，如果獲取鎖成功就返回，否則就阻塞。直到上一個(gè)節(jié)點(diǎn)釋放鎖并喚醒它。這樣看來(lái)似乎跟自旋沒(méi)什么掛鉤。這是因?yàn)锳QS里面的CLH隊(duì)列是CLH隊(duì)列鎖的一種變形。先來(lái)了解一下CLH隊(duì)列鎖
CLH隊(duì)列鎖
CLH(Craig, Landin, and Hagersten locks): 是一個(gè)自旋鎖，能確保無(wú)饑餓性，提供先來(lái)先服務(wù)的公平性。
CLH鎖也是一種基于鏈表的可擴(kuò)展、高性能、公平的自旋鎖，申請(qǐng)線程只在本地變量上自旋，它不斷輪詢前驅(qū)的狀態(tài)，如果發(fā)現(xiàn)前驅(qū)釋放了鎖就結(jié)束自旋。http://www.2cto.com/kf/201412/363574.html這篇文章中有比較詳細(xì)的圖解。
AQS中的CLH隊(duì)列
了解了自旋鎖與CLH隊(duì)列鎖之后，在學(xué)習(xí)AQS中的CLH隊(duì)列就比較簡(jiǎn)單了。AQS中的CLH隊(duì)列主要是對(duì)CLH隊(duì)列鎖改動(dòng)了兩個(gè)地方
1.節(jié)點(diǎn)結(jié)構(gòu)上做出改變。CLH隊(duì)列鎖的節(jié)點(diǎn)包含一個(gè)布爾類型locked的字段。如果要獲取鎖，就將這個(gè)locked設(shè)置為true。然后就不停的輪訓(xùn)前驅(qū)節(jié)點(diǎn)的locked是否釋放了鎖（這個(gè)過(guò)程我們就叫做自旋）。AQS的CLH隊(duì)列在結(jié)構(gòu)上引入了頭節(jié)點(diǎn)，尾節(jié)點(diǎn)。并且擁有一個(gè)前節(jié)點(diǎn)與下一個(gè)節(jié)點(diǎn)的引用。
2.在等待獲取鎖的機(jī)制上由自旋改成了等待阻塞。
MCS
MSC與CLH最大的不同并不是鏈表是顯示還是隱式，而是線程自旋的規(guī)則不同:CLH是在前趨結(jié)點(diǎn)的locked域上自旋等待，而MSC是在自己的
結(jié)點(diǎn)的locked域上自旋等待。正因?yàn)槿绱?，它解決了CLH在NUMA系統(tǒng)架構(gòu)中獲取locked域狀態(tài)內(nèi)存過(guò)遠(yuǎn)的問(wèn)題。
重慶分公司
重慶分公司
/*
* 提供了一個(gè)基于FIFO隊(duì)列，可以用于構(gòu)建鎖或者其他相關(guān)同步裝置的基礎(chǔ)框架
* 雙向鏈表
*/
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
/**
* 無(wú)參構(gòu)造方法
*/
protected AbstractQueuedSynchronizer() { }
/**
*
* +------+ prev +-----+ +-----+
* head | | <---- | | <---- | | tail
* +------+ +-----+ +-----+
*
*/
static final class Node {
/** Marker to indicate a node is waiting in shared mode 模式，分為共享與獨(dú)占 共享模式 */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode 獨(dú)占模式 */
static final Node EXCLUSIVE = null;
/** waitStatus value to indicate thread has cancelled
* 結(jié)點(diǎn)狀態(tài) 節(jié)點(diǎn)watiStatus的值
* CANCELLED，值為1，終態(tài)，該節(jié)點(diǎn)被取消由于超時(shí)或中斷
* SIGNAL，值為-1，表示當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)包含的線程需要運(yùn)行，也就是unpark，所以當(dāng)前節(jié)點(diǎn)release或cancels時(shí)，必須unpark它的后繼節(jié)點(diǎn)
* CONDITION，值為-2，表示當(dāng)前節(jié)點(diǎn)在等待condition，也就是在condition隊(duì)列中 該節(jié)點(diǎn)處于條件隊(duì)列中，將不會(huì)被用于sync queue，直到節(jié)點(diǎn)狀態(tài)被設(shè)置為0
* PROPAGATE，值為-3，表示當(dāng)前場(chǎng)景下后續(xù)的acquireShared能夠得以執(zhí)行releaseShared應(yīng)該被傳播到其他節(jié)點(diǎn)
* 值為0，表示當(dāng)前節(jié)點(diǎn)在sync隊(duì)列中，等待著獲取鎖
* */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;
/**
* Status field, taking on only the values:
* SIGNAL: The successor of this node is (or will soon be)
* blocked (via park), so the current node must
* unpark its successor when it releases or
* cancels. To avoid races, acquire methods must
* first indicate they need a signal,
* then retry the atomic acquire, and then,
* on failure, block.
* CANCELLED: This node is cancelled due to timeout or interrupt.
* Nodes never leave this state. In particular,
* a thread with cancelled node never again blocks.
* CONDITION: This node is currently on a condition queue.
* It will not be used as a sync queue node
* until transferred, at which time the status
* will be set to 0. (Use of this value here has
* nothing to do with the other uses of the
* field, but simplifies mechanics.)
* PROPAGATE: A releaseShared should be propagated to other
* nodes. This is set (for head node only) in
* doReleaseShared to ensure propagation
* continues, even if other operations have
* since intervened.
* 0: None of the above
* 結(jié)點(diǎn)狀態(tài)
*/
volatile int waitStatus;
/**
* 前驅(qū)結(jié)點(diǎn)
*/
volatile Node prev;
/**
* 后繼結(jié)點(diǎn)
*/
volatile Node next;
/**
* 結(jié)點(diǎn)所對(duì)應(yīng)的線程
*/
volatile Thread thread;
/**
* 下一個(gè)等待者
*/
Node nextWaiter;
/**
* 結(jié)點(diǎn)是否在共享模式下等待
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* 獲取前驅(qū)結(jié)點(diǎn)，若前驅(qū)結(jié)點(diǎn)為空，拋出異常
*/
final Node predecessor() throws NullPointerException {
// 保存前驅(qū)結(jié)點(diǎn)
Node p = prev;
if (p == null) // 前驅(qū)結(jié)點(diǎn)為空，拋出異常
throw new NullPointerException();
else // 前驅(qū)結(jié)點(diǎn)不為空，返回
return p;
}
// 無(wú)參構(gòu)造函數(shù)
Node() { // Used to establish initial head or SHARED marker
}
// 構(gòu)造函數(shù)
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
// 構(gòu)造函數(shù)
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
/**
* CLH隊(duì)列中頭結(jié)點(diǎn)
*/
private transient volatile Node head;
/**
* CLH隊(duì)列中尾結(jié)點(diǎn)
*/
private transient volatile Node tail;
/**
* 同步狀態(tài)
* 多線程同步獲取資源成功，則state字段會(huì)自增；若有線程釋放資源，則state字段自減。
* 信號(hào)量 記錄該線程持有鎖的次數(shù)。 該線程每次釋放所 信號(hào)量 -1。 信號(hào)量為零 代表 鎖被真正釋放
*/
private volatile int state;
/**
* @return current state value
*/
protected final int getState() {
return state;
}
/**
* @param newState the new state value
*/
protected final void setState(int newState) {
state = newState;
}
/**
* 使用unsafe的cas比較并且交換，保證原子性
*/
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
?? AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer。
? 1. AbstractOwnableSynchronizer是一個(gè)抽象類一個(gè)同步器，它可能由線程獨(dú)占。該類為創(chuàng)建可能包含所有權(quán)概念的鎖和相關(guān)同步器提供了基礎(chǔ)，但是，子類和工具可以使用適當(dāng)維護(hù)的值來(lái)幫助控制和監(jiān)視訪問(wèn)并提供診斷，實(shí)現(xiàn)了Serializable接口，定義了獨(dú)占模式，設(shè)置和獲取獨(dú)占模式下的線程Thread信息。
? 2.AbstractOwnableSynchronizer實(shí)現(xiàn)了Serializable接口。
? ??1)Serializable接口，序列化接口，表明該類可以被序列化，什么是序列化？簡(jiǎn)單的說(shuō)，就是能夠從類變成字節(jié)流傳輸，反序列化，就是從字節(jié)流變成原來(lái)的類。
? 3. AbstractOwnableSynchronizer是一個(gè)抽象父類，子類有AbstractQueuedSynchronizer和AbstractQueuedLongSynchronizer，它們2個(gè)之間的區(qū)別就是異常將所有與狀態(tài)相關(guān)的參數(shù)和結(jié)果定義為long類型而不是int類型，在創(chuàng)建同步器（例如多級(jí)鎖和需要64位狀態(tài)的障礙）時(shí)，此類可能很有用。??
?? ??1)acquire(int arg)；
? ??以獨(dú)占模式獲取資源，如果獲取成功，直接返回，否則進(jìn)去CLH等待隊(duì)列，通過(guò)自旋知道獲取到資源為止，過(guò)程中忽略線程中斷，獲取資源后才進(jìn)行自我中斷(補(bǔ)上)，下面看源碼：
/**
* AQS的獨(dú)占模式--互斥
* tryAcquire()嘗試直接去獲取資源，如果成功則直接返回；
* addWaiter()將該線程加入等待隊(duì)列的尾部，并標(biāo)記為獨(dú)占模式；
* acquireQueued()使線程在等待隊(duì)列中獲取資源，一直獲取到資源后才返回。如果在整個(gè)等待過(guò)程中被中斷過(guò)，則返回true，否則返回false。
* 如果線程在等待過(guò)程中被中斷過(guò)，它是不響應(yīng)的。只是獲取資源后才再進(jìn)行自我中斷selfInterrupt()，將中斷補(bǔ)上。
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) && // 再次嘗試上鎖 回到了 NonfairSync.tryAcquire 方法， tryAcquire 調(diào)用了 Sync.nonfairTryAcquire方法
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 鏈表尾部添加節(jié)點(diǎn) 為獨(dú)占模式
selfInterrupt();
}?? ??2)boolean tryAcquire(int arg)；
? ??嘗試以獨(dú)占的方式獲取資源，成功true，失敗false，該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
/**
* tryAcquire嘗試以獨(dú)占的方式獲取資源，如果獲取成功，則直接返回true，否則直接返回false。該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
* 該方法的默認(rèn)實(shí)現(xiàn)是拋出UnsupportedOperationException，具體實(shí)現(xiàn)由自定義的擴(kuò)展了AQS的同步類來(lái)實(shí)現(xiàn)。AQS在這里只負(fù)責(zé)定義了一個(gè)公共的方法框架。
* 這里之所以沒(méi)有定義成abstract，是因?yàn)楠?dú)占模式下只用實(shí)現(xiàn)tryAcquire-tryRelease，而共享模式下只用實(shí)現(xiàn)tryAcquireShared-tryReleaseShared。
* 如果都定義成abstract，那么每個(gè)模式也要去實(shí)現(xiàn)另一模式下的接口
* 由子類選擇性實(shí)現(xiàn)
*/
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}?? ??3)Node addWaiter(Node mode)；
? ??將一個(gè)Node節(jié)點(diǎn)放入到CLH隊(duì)列的隊(duì)尾。
/**
* 將一個(gè)Node節(jié)點(diǎn)放入到CLH隊(duì)列的隊(duì)尾。
* 第一步：首先將oldTail賦值給newNode.prev：node.prev = pred, 把當(dāng)前tail節(jié)點(diǎn)賦值到mode新節(jié)點(diǎn)的prev前一個(gè),
* 第二步：將tail賦值給newNode：compareAndSetTail(pred, node) 把當(dāng)前tail節(jié)點(diǎn)的內(nèi)存地址修改為(指向)新的mode節(jié)點(diǎn)，
* 第三步：將oldTail的next指針指向newNode(即tail)：pred.next = node 把當(dāng)前tail節(jié)點(diǎn)的next后一個(gè)賦值為新的mode節(jié)點(diǎn)(即tail)
* 如果隊(duì)列為空，通過(guò)enq(node)方法初始化一個(gè)等待隊(duì)列，并返回當(dāng)前節(jié)點(diǎn)
*/
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
//嘗試快速入隊(duì)，失敗則使用enq()方式
Node pred = tail;
if (pred != null) { // 列隊(duì)尾部不為空
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 列隊(duì)尾部為空 或者 CAS 操作失敗
enq(node);
return node;
}?? ??4)boolean acquireQueued(final Node node, int arg)；
? ??使線程在等待隊(duì)列中獲取資源，一直獲取到資源后才返回。如果在整個(gè)等待過(guò)程中被中斷過(guò)，則返回true，否則返回false。
/**
* 若node節(jié)點(diǎn)的前繼節(jié)點(diǎn)是head節(jié)點(diǎn)，則會(huì)再次調(diào)用tryAcquire()獲取資源
* 判斷當(dāng)前節(jié)點(diǎn)的前繼節(jié)點(diǎn)是否為head節(jié)點(diǎn)。若是，則表示該節(jié)點(diǎn)有資格嘗試獲取共享資源。此處的head節(jié)點(diǎn)的判斷在一定程度上保證資源競(jìng)爭(zhēng)的公平性
* shouldParkAfterFailedAcquire()：判斷當(dāng)前節(jié)點(diǎn)是否可以安全進(jìn)入park()
* parkAndCheckInterrupt()：讓線程進(jìn)入等待
* 用于隊(duì)列中的線程自旋地以獨(dú)占且不可中斷的方式獲取同步狀態(tài)（acquire），直到拿到鎖之后再返回。該方法的實(shí)現(xiàn)分成兩部分：
* 如果當(dāng)前節(jié)點(diǎn)已經(jīng)成為頭結(jié)點(diǎn)，嘗試獲取鎖（tryAcquire）成功，然后返回；否則檢查當(dāng)前節(jié)點(diǎn)是否應(yīng)該被park，然后將該線程park并且檢查當(dāng)前線程是否被可以被中斷
*/
final boolean acquireQueued(final Node node, int arg) {
//標(biāo)記是否成功拿到資源，默認(rèn)false
boolean failed = true;
try {
boolean interrupted = false;//標(biāo)記等待過(guò)程中是否被中斷過(guò)
for (;;) {
final Node p = node.predecessor();
// 判斷當(dāng)前節(jié)點(diǎn)的 前驅(qū)節(jié)點(diǎn) 是否為隊(duì)列頭部 如果是 再次嘗試上鎖（如果頭部節(jié)點(diǎn) 已經(jīng)釋放鎖, 則使當(dāng)前線程成為持有者 并且設(shè)置自己為 頭部。 同時(shí)釋放前驅(qū)節(jié)點(diǎn)）
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//判斷當(dāng)前節(jié)點(diǎn)是否可以進(jìn)入park，若可以，讓線程進(jìn)入等待
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
//如果獲取資源失敗，則取消
if (failed)
cancelAcquire(node);
}
}?? ??5)void selfInterrupt();
? ??中斷當(dāng)前線程
/**
* Convenience method to interrupt current thread.
* 中斷當(dāng)前線程
*/
static void selfInterrupt() {
Thread.currentThread().interrupt();
}?? ??6)Node enq(final Node node)；
? ??將當(dāng)前節(jié)點(diǎn)插入等待隊(duì)列
/**
* 進(jìn)行自旋入隊(duì)方式的enq()方法，基本和addWaiter()方法一致：
* 用于將當(dāng)前節(jié)點(diǎn)插入等待隊(duì)列，如果隊(duì)列為空，則初始化當(dāng)前隊(duì)列。整個(gè)過(guò)程以CAS自旋的方式進(jìn)行，直到成功加入隊(duì)尾為止
*/
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize 必須初始化 尾部為空 嘗試構(gòu)建表結(jié)構(gòu)
if (compareAndSetHead(new Node()))
tail = head;
} else { //尾部不為空 不斷嘗試 CAS 操作
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}?? ??7)boolean compareAndSetHead(Node update)；
? ??通過(guò)原子(CAS)操作 改變上鎖狀態(tài)
/**
* CAS head field. Used only by enq.
* 通過(guò)原子操作 改變上鎖狀態(tài)
* this == null
* 第一個(gè)參數(shù)為需要改變的對(duì)象，第二個(gè)為偏移量(即之前求出來(lái)的valueOffset的值)，第三個(gè)參數(shù)為期待的值，第四個(gè)為更新后的值
*/
private final boolean compareAndSetHead(Node update) {
//調(diào)用本地方法 實(shí)現(xiàn)硬件級(jí)別的原子操作 cas
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}?? ??8)boolean parkAndCheckInterrupt()；
? ??讓線程去休息，真正進(jìn)入等待狀態(tài)
?
/**
* 該方法讓線程去休息，真正進(jìn)入等待狀態(tài)。park()會(huì)讓當(dāng)前線程進(jìn)入waiting狀態(tài)。在此狀態(tài)下，有兩種途徑可以喚醒該線程：
* 1）被unpark()；2）被interrupt()。需要注意的是，Thread.interrupted()會(huì)清除當(dāng)前線程的中斷標(biāo)記位
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 又是一個(gè)底層類 實(shí)現(xiàn)線程等待
return Thread.interrupted();
}?? ??9)boolean shouldParkAfterFailedAcquire(Node pred, Node node)；
? ??判斷當(dāng)前節(jié)點(diǎn)中的線程，是否可以安全的進(jìn)入park()。返回true,表示進(jìn)程可以進(jìn)入park
/**
* 該方法的作用在于判斷當(dāng)前節(jié)點(diǎn)中的線程，是否可以安全的進(jìn)入park()。返回true,表示進(jìn)程可以進(jìn)入park。若前驅(qū)節(jié)點(diǎn)的waitStatus為SIGNAL，則表示當(dāng)前節(jié)點(diǎn)可以安全的park()。
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//如果前驅(qū)節(jié)點(diǎn)的waitStatus為SIGNAL -1，則表示當(dāng)前節(jié)點(diǎn)可以安全的park()
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
// waitStatus>0，即為CANCELLED狀態(tài)，此時(shí)當(dāng)前節(jié)點(diǎn)需要找到狀態(tài)不為CANCELLED狀態(tài)的節(jié)點(diǎn)，將其設(shè)置為自己的前驅(qū)節(jié)點(diǎn)，并將新的前驅(qū)節(jié)點(diǎn)的next指向自己。
// 注意，這樣做完之后，那些當(dāng)前節(jié)點(diǎn)的waitStatus狀態(tài)為CANCELLED的前驅(qū)節(jié)點(diǎn)鏈，將成為孤鏈。但這個(gè)孤鏈仍然有指向原等待隊(duì)列的prev和next指針。只是原等待隊(duì)列中已經(jīng)沒(méi)有指向孤鏈的節(jié)點(diǎn)指針
// 將前驅(qū)節(jié)點(diǎn)移出列隊(duì)
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* 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.
*/
// 走到此處，表明前驅(qū)節(jié)點(diǎn)的狀態(tài)為0或PROPAGATE。此時(shí)可以將前驅(qū)節(jié)點(diǎn)的waitStatus設(shè)置為SIGNAL狀態(tài)
// 注意：這里仍然要返回false，表明當(dāng)前節(jié)點(diǎn)不能被park。我們需要在park之前，重試確認(rèn)該節(jié)點(diǎn)不能獲取到資源
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
} ?? ??1)acquireShared(int arg)；
? ??以共享模式獲取資源，如果獲取成功，直接返回，否則進(jìn)去CLH等待隊(duì)列，通過(guò)自旋知道獲取到資源為止，過(guò)程中忽略線程中斷，獲取資源后才進(jìn)行自我中斷(補(bǔ)上)，下面看源碼：
/**
* aqs的共享模式
* 獲取指定量的資源，獲取成功則直接返回，獲取失敗則進(jìn)入等待隊(duì)列，直到獲取到資源為止，整個(gè)過(guò)程忽略中斷
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}?? ??2)int tryAcquireShared(int arg)；
? ??嘗試以共享的方式獲取資源，成功true，失敗false，該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
/**
* tryAcquireShared嘗試以共享的方式獲取資源，如果獲取成功，則直接返回true，否則直接返回false。該方法可以用于實(shí)現(xiàn)Lock中的tryLock()方法。
* 該方法的默認(rèn)實(shí)現(xiàn)是拋出UnsupportedOperationException，具體實(shí)現(xiàn)由自定義的擴(kuò)展了AQS的同步類來(lái)實(shí)現(xiàn)。AQS在這里只負(fù)責(zé)定義了一個(gè)公共的方法框架。
* 這里之所以沒(méi)有定義成abstract，是因?yàn)楠?dú)占模式下只用實(shí)現(xiàn)tryAcquire-tryRelease，而共享模式下只用實(shí)現(xiàn)tryAcquireShared-tryReleaseShared。
* 如果都定義成abstract，那么每個(gè)模式也要去實(shí)現(xiàn)另一模式下的接口
*/
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}?? ??3)doAcquireShared(int arg)；
? ??將當(dāng)前線程加入等待隊(duì)列尾部休息，直到其他線程釋放資源喚醒自己，自己成功拿到相應(yīng)量的資源后才返回
/**
* Acquires in shared uninterruptible mode.
* @param arg the acquire argument
* 將當(dāng)前線程加入等待隊(duì)列尾部休息，直到其他線程釋放資源喚醒自己，自己成功拿到相應(yīng)量的資源后才返回
*/
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}?? ??4)boolean parkAndCheckInterrupt()；
? ??讓線程去休息，真正進(jìn)入等待狀態(tài)
/**
* 該方法讓線程去休息，真正進(jìn)入等待狀態(tài)。park()會(huì)讓當(dāng)前線程進(jìn)入waiting狀態(tài)。在此狀態(tài)下，有兩種途徑可以喚醒該線程：
* 1）被unpark()；2）被interrupt()。需要注意的是，Thread.interrupted()會(huì)清除當(dāng)前線程的中斷標(biāo)記位
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 又是一個(gè)底層類 實(shí)現(xiàn)線程等待
return Thread.interrupted();
}?? ??5)cancelAcquire(Node node)；
? ??取消節(jié)點(diǎn)
/**
* 取消節(jié)點(diǎn)
* 列隊(duì)等待中 拋出異常會(huì)調(diào)用此方法
*/
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
//找到適合的前繼節(jié)點(diǎn)，當(dāng)前節(jié)點(diǎn)的waitStatus賦值為CANCELLED
node.thread = null; // 釋放線程
// Skip cancelled predecessors 前驅(qū)節(jié)點(diǎn)已被取消 重新定義前驅(qū)節(jié)點(diǎn)
Node pred = node.prev;
//若前繼節(jié)點(diǎn)是CANCELLED，則繼續(xù)找前繼節(jié)點(diǎn)，直至找到一個(gè)正常的前繼節(jié)點(diǎn)賦值給node，作為node的新前繼節(jié)點(diǎn)
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
Node predNext = pred.next;
node.waitStatus = Node.CANCELLED; // 取消當(dāng)前線程 所屬的節(jié)點(diǎn)（標(biāo)記為取消）， 沒(méi)有使用 cas 因?yàn)?其他線程 不會(huì)干擾這里
// If we are the tail, remove ourselves.
//特殊情況：node==tail節(jié)點(diǎn)，將pred作為tail節(jié)點(diǎn)，然后將cancelledNodes節(jié)點(diǎn)鏈從CLH隊(duì)列剔除
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
//正常情況：則將cancelledNodes節(jié)點(diǎn)鏈從CLH隊(duì)列剔除
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
//特殊情況：如果node是head的后繼節(jié)點(diǎn)，則直接喚醒node的后繼節(jié)點(diǎn) pred==head節(jié)點(diǎn)：嘗試調(diào)用unparkSuccessor(node)，嘗試喚醒當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)
unparkSuccessor(node);
}
node.next = node; // help GC
}
}?? ??6)unparkSuccessor(Node node)；
? ??喚醒后繼節(jié)點(diǎn)
/**
* 喚醒后繼節(jié)點(diǎn)
* 注意：如果當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)為空，或者是被取消的節(jié)點(diǎn)。那就從tail節(jié)點(diǎn)逆向遍歷CLH隊(duì)列，直至找到一個(gè)距離當(dāng)前節(jié)點(diǎn)node最近，且waitStatus<=0的節(jié)點(diǎn)，然后喚醒該節(jié)點(diǎn)
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus; //獲取頭結(jié)點(diǎn)的等待狀態(tài)
if (ws < 0) //把該狀態(tài)設(shè)置成0
compareAndSetWaitStatus(node, ws, 0);
/*
* 若后繼節(jié)點(diǎn)不符合喚醒標(biāo)準(zhǔn)，則逆向遍歷CLH，直至找到一個(gè)距離當(dāng)前節(jié)點(diǎn)node最近，且waitStatus<=0的節(jié)點(diǎn)
*/
Node s = node.next; //找到后繼節(jié)點(diǎn),喚醒后繼節(jié)點(diǎn)
if (s == null || s.waitStatus > 0) { //很不巧，后繼節(jié)點(diǎn)，節(jié)點(diǎn)為null,或者被取消
s = null;
for (Node t = tail; t != null && t != node; t = t.prev) //這里采用反向遍歷因?yàn)槭请p向鏈表
if (t.waitStatus <= 0) //找到實(shí)際未被取消的節(jié)點(diǎn)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread); //喚醒節(jié)點(diǎn)
}?? ??1)boolean release(int arg)；
? ??獨(dú)占模式釋放資源
/**
* 資源的釋放
* 調(diào)用tryRelease方法進(jìn)行釋放鎖
* 釋放鎖成功后，獲取頭節(jié)點(diǎn)，接著喚醒后繼節(jié)點(diǎn)，調(diào)用unparkSuccessor方法
*/
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}?? ??2)boolean tryRelease(int arg)；
? ??獨(dú)占模式下嘗試釋放鎖，由子類選擇性實(shí)現(xiàn)
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}?? ??3)boolean releaseShared(int arg)；
? ??共享模式釋放資源
/**
* Releases in shared mode. Implemented by unblocking one or more
* threads if {@link #tryReleaseShared} returns true.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryReleaseShared} but is otherwise uninterpreted
* and can represent anything you like.
* @return the value returned from {@link #tryReleaseShared}
*/
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}?? ??4)boolean tryReleaseShared(int arg)；
? ??共享模式下嘗試釋放鎖，由子類選擇性實(shí)現(xiàn)
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}?? ??5)int fullyRelease(Node node)；
? ??使用當(dāng)前節(jié)點(diǎn)狀態(tài)調(diào)用release,成功返回狀態(tài)，失敗跑出異常
/**
* 使用當(dāng)前節(jié)點(diǎn)狀態(tài)調(diào)用release,成功返回狀態(tài)，失敗跑出異常
*/
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
} /**
* Unsafe類實(shí)例
*/
private static final Unsafe unsafe = Unsafe.getUnsafe();
/** state內(nèi)存偏移地址 */
private static final long stateOffset;
/** head內(nèi)存偏移地址 */
private static final long headOffset;
/** tail內(nèi)存偏移地址 */
private static final long tailOffset;
private static final long waitStatusOffset;
/** next內(nèi)存偏移地址 */
private static final long nextOffset;
//靜態(tài)初始化塊
static {
try {
// 獲取偏移量
stateOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
waitStatusOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("waitStatus"));
nextOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("next"));
} catch (Exception ex) { throw new Error(ex); }
}
/**
* CAS head field. Used only by enq.
* 通過(guò)原子操作 改變上鎖狀態(tài)
* this == null
* 第一個(gè)參數(shù)為需要改變的對(duì)象，第二個(gè)為偏移量(即之前求出來(lái)的valueOffset的值)，第三個(gè)參數(shù)為期待的值，第四個(gè)為更新后的值
*/
private final boolean compareAndSetHead(Node update) {
//調(diào)用本地方法 實(shí)現(xiàn)硬件級(jí)別的原子操作 cas
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
/**
* CAS tail field. Used only by enq.
*/
private final boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
/**
* CAS waitStatus field of a node.
*/
private static final boolean compareAndSetWaitStatus(Node node,
int expect,
int update) {
return unsafe.compareAndSwapInt(node, waitStatusOffset,
expect, update);
}
/**
* CAS next field of a node.
*/
private static final boolean compareAndSetNext(Node node,
Node expect,
Node update) {
return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
}內(nèi)部類ConditionObject，它實(shí)現(xiàn)了Condition接口，主要用于實(shí)現(xiàn)條件鎖。
ConditionObject中也維護(hù)了一個(gè)隊(duì)列，這個(gè)隊(duì)列主要用于等待條件的成立，當(dāng)條件成立時(shí)，其它線程將signal這個(gè)隊(duì)列中的元素，將其移動(dòng)到CLH的隊(duì)列中，等待占有鎖的線程釋放鎖后被喚醒。
Condition典型的運(yùn)用場(chǎng)景是在BlockingQueue中的實(shí)現(xiàn)，當(dāng)隊(duì)列為空時(shí)，獲取元素的線程阻塞在notEmpty條件上，一旦隊(duì)列中添加了一個(gè)元素，將通知notEmpty條件，將其隊(duì)列中的元素移動(dòng)到AQS隊(duì)列中等待被喚醒。
/**
* 構(gòu)造一個(gè)條件隊(duì)列，來(lái)等待條件是否為真
*/
public class ConditionObject implements Condition, java.io.Serializable {
/** 版本號(hào) */
private static final long serialVersionUID = 1173984872572414699L;
/** First node of condition queue. condition隊(duì)列的頭結(jié)點(diǎn) */
private transient Node firstWaiter;
/** Last node of condition queue. condition隊(duì)列的尾結(jié)點(diǎn) */
private transient Node lastWaiter;
/**
* Creates a new {@code ConditionObject} instance.
* 構(gòu)造函數(shù)
*/
public ConditionObject() { }
/**
* Adds a new waiter to wait queue.
* @return its new wait node
* 添加新的waiter到wait隊(duì)列
*/
private Node addConditionWaiter() {
//保存尾結(jié)點(diǎn)
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out. 尾結(jié)點(diǎn)不為空，并且尾結(jié)點(diǎn)的狀態(tài)不為CONDITION
if (t != null && t.waitStatus != Node.CONDITION) {
//清除狀態(tài)為CONDITION的結(jié)點(diǎn)
unlinkCancelledWaiters();
//將最后一個(gè)結(jié)點(diǎn)重新賦值給t
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}1）AQS分為獨(dú)占鎖和共享鎖。
2）AQS分為CLH自旋隊(duì)列和Condition條件隊(duì)列。
3）AQS是一個(gè)雙向鏈表，由state狀態(tài)控制。
4）AQS由volatile修飾保證多線程可見(jiàn)，采用CAS保證原子性。
