AQS是個(gè)啥？AQS(AbstractQueuedSynchronizer)是Java并發(fā)用來(lái)構(gòu)建鎖和其他同步組件的基礎(chǔ)框架。許多同步類實(shí)現(xiàn)都依賴于它，如常用的ReentrantLock/ReentrantReadWriterLock/CountDownLatch等 AQS提供了獨(dú)占(Exclusive)以及共享(Share)兩種資源共享方式： acquire(acquireShare)/release(releaseShare)。 acquire：獲取資源，如果當(dāng)前資源滿足條件，則直接返回，否則掛起當(dāng)前線程，將該線程加入到隊(duì)列排隊(duì)。 release：釋放資源，喚醒掛起線程AQS隊(duì)列AQS隊(duì)列示意圖
AQS隊(duì)列中的主要屬性// 等待隊(duì)列頭部 private transient volatile Node head; // 等待隊(duì)列尾部 private transient volatile Node tail; // 鎖的狀態(tài)（加鎖成功則為1，解鎖為0，重入再 1） private volatile int state; // 當(dāng)前持有鎖的線程，注意這個(gè)屬性是從AbstractOwnableSynchronizer繼承而來(lái) private transient Thread exclusiveOwnerThread; Node類中的主要屬性static final class Node {
// 標(biāo)記表示節(jié)點(diǎn)正在共享模式中等待
static final Node SHARED = new Node();
// 標(biāo)記表示節(jié)點(diǎn)正在獨(dú)占模式下等待
static final Node EXCLUSIVE = null;
// 節(jié)點(diǎn)的等待狀態(tài) 還有一個(gè)初始化狀態(tài)0 不屬于以下四種狀態(tài)
// 表示Node所代表的當(dāng)前線程已經(jīng)取消了排隊(duì)，即放棄獲取鎖
static final int CANCELLED = 1;
// 當(dāng)一個(gè)節(jié)點(diǎn)的waitStatus被置為SIGNAL，就說(shuō)明它的下一個(gè)節(jié)點(diǎn)（即它的后繼節(jié)點(diǎn)）已經(jīng)被掛起了（或者馬上就要被掛起了），
// 只要前繼結(jié)點(diǎn)釋放鎖，就會(huì)通知標(biāo)識(shí)為SIGNAL狀態(tài)的后繼結(jié)點(diǎn)的線程執(zhí)行
static final int SIGNAL = -1;
// 節(jié)點(diǎn)在等待隊(duì)列中
// 當(dāng)其他線程對(duì)Condition調(diào)用了signal()后，該節(jié)點(diǎn)將會(huì)從等待隊(duì)列中轉(zhuǎn)移到同步隊(duì)列中，加入到同步狀態(tài)的獲取中
static final int CONDITION = -2;
// 表示下一次共享式同步狀態(tài)獲取，將會(huì)無(wú)條件地傳播下去
static final int PROPAGATE = -3;
// 節(jié)點(diǎn)等待狀態(tài)，該字段初始化為0，
volatile int waitStatus;
// 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)
volatile Node prev;
// 當(dāng)前節(jié)點(diǎn)的后置節(jié)點(diǎn)
volatile Node next;
// 在此節(jié)點(diǎn)上排隊(duì)的線程信息
volatile Thread thread;
}ReentrantLock實(shí)現(xiàn)在引入ReentrantLock實(shí)現(xiàn)前，我先來(lái)科普一下 util.concurrent包的作者Doug Lea，相比較其他而言，并發(fā)包的源碼閱讀難度較大。臉上永遠(yuǎn)掛著謙遜靦腆笑容的Doug Lea先生使用了大量相對(duì)復(fù)雜的邏輯判斷，比如一個(gè)判斷條件中執(zhí)行多個(gè)或且方法，讓你很難跟上他的節(jié)奏，很難揣摩他的設(shè)計(jì)思想。小聲逼逼，還不是我太菜了，留下來(lái)沒(méi)有技術(shù)的淚水。 繼承關(guān)系圖ReentrantLock是Lock接口的一個(gè)實(shí)現(xiàn)類，是一種可重入的獨(dú)占鎖。 ReentrantLock內(nèi)部通過(guò)內(nèi)部類實(shí)現(xiàn)了AQS框架(AbstractQueuedSynchronizer)的API來(lái)實(shí)現(xiàn)獨(dú)占鎖的功能。 主要屬性private final Sync sync;
// 公平鎖內(nèi)部是FairSync,非公平鎖內(nèi)部是NonfairSync。
// 兩者都通過(guò)繼承 Sync間接繼承自AbstractQueuedSynchronizer這個(gè)抽象類
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
// 加鎖
abstract void lock();
// 嘗試獲取鎖
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
// 嘗試釋放鎖
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
}構(gòu)造方法//默認(rèn)創(chuàng)建一個(gè)非公平鎖
public ReentrantLock() {
sync = new NonfairSync();
}
//傳入true創(chuàng)建公平鎖，false非公平鎖
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}ReentrantLock公平鎖我們以公平鎖為例對(duì)其中重要方法源碼分析 // 繼承了 Sync，從而間接繼承了 AbstractQueuedSynchronizer這個(gè)抽象類
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
// 上鎖
final void lock() {
//調(diào)用 AQS 中 acquire方法
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// CAS操作設(shè)置 state
// 設(shè)置當(dāng)前線程為擁有鎖的線程
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}acquire方法源碼分析public final void acquire(int arg) {
// tryAcquire(arg)嘗試加鎖，如果加鎖失敗則會(huì)調(diào)用acquireQueued方法加入隊(duì)列去排隊(duì)，如果加鎖成功則不會(huì)調(diào)用
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}acquire方法干了這么幾件事情
1、tryAcquire() 嘗試獲取資源，如果成功則直接返回；
2、addWaiter() 將該線程加入等待隊(duì)列， 更新AQS隊(duì)列鏈信息
3、acquireQueued() 使線程在等待隊(duì)列中獲取資源，直到獲取資源后才返回。如果在整個(gè)等待過(guò)程中被中斷過(guò)，則返回true，否則返回false。
4、selfInterrupt() 自我中斷，如果線程在等待過(guò)程中被中斷過(guò)，它是不響應(yīng)的。只是獲取資源后再將中斷補(bǔ)上。
tryAcquire方法protected final boolean tryAcquire(int acquires) {
// 獲取當(dāng)前線程
final Thread current = Thread.currentThread();
// 獲取lock對(duì)象的上鎖狀態(tài)，如果鎖是自由狀態(tài)則=0，如果被上鎖則為1，大于1表示重入
int c = getState();
// c=0 代表沒(méi)人占用鎖，當(dāng)前線程可以直接獲取鎖資源執(zhí)行
if (c == 0) {
// 下面介紹hasQueuedPredecessors()方法，判斷自己是否需要排隊(duì)
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
// CAS操作設(shè)置 state
// 設(shè)置當(dāng)前線程為擁有鎖的線程
setExclusiveOwnerThread(current);
return true;
}
}
// 非重入鎖直接返回false，加鎖失敗
else if (current == getExclusiveOwnerThread()) {
// 若為重入鎖, state 加1 (acquires)
int nextc = c acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}hasQueuedPredecessors方法public final boolean hasQueuedPredecessors() {
// 獲取隊(duì)列頭、尾節(jié)點(diǎn)信息
Node t = tail;
Node h = head;
Node s;
// h != t 有幾種情況
// 1、隊(duì)列尚未初始化完成，第一個(gè)線程獲取鎖資源，
// 此時(shí)h和t都是null, h != t返回fasle初始化隊(duì)列
// 2、隊(duì)列已經(jīng)被初始化了，其他的線程嘗試獲取資源，
// 此時(shí)頭尾節(jié)點(diǎn)不相同，h!=t返回true，
// 繼續(xù)判斷s.thread != Thread.currentThread() 當(dāng)前來(lái)參與競(jìng)爭(zhēng)鎖的線程和第一個(gè)排隊(duì)的線程是同一個(gè)線程,則需要排隊(duì)。
// 3、隊(duì)列已經(jīng)被初始化了，但是由于鎖釋放的原因?qū)е玛?duì)列里面只有一個(gè)數(shù)據(jù)
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}addWaiter方法private Node addWaiter(Node mode) {
// AQS隊(duì)列中的元素類型為Node，需要把當(dāng)前線程封裝成為一個(gè)Node對(duì)象
Node node = new Node(Thread.currentThread(), mode);
// tail為隊(duì)尾，賦值給pred
Node pred = tai
// 判斷pred是否為空，其實(shí)就是判斷隊(duì)尾是否有節(jié)點(diǎn)，其實(shí)只要隊(duì)列被初始化了隊(duì)尾肯定不為空，
if (pred != null) {
// 拼裝node隊(duì)列鏈的過(guò)程
// 直接把當(dāng)前線程封裝的node的上一個(gè)節(jié)點(diǎn)設(shè)置成為pred即原來(lái)的隊(duì)尾
node.prev = pred;
if (compareAndSetTail(pred, node)) {
// pred的下一個(gè)節(jié)點(diǎn)設(shè)置為當(dāng)node
pred.next = node;
return node;
}
}
// 拼接aqs隊(duì)列鏈
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}acquireQueued方法final boolean acquireQueued(final Node node, int arg) {
// 標(biāo)記是否成功拿到資源
boolean failed = true;
try {
// 標(biāo)記等待過(guò)程中是否被中斷過(guò)
boolean interrupted = false;
// 自旋
for (;;) {
final Node p = node.predecessor();
// 判斷自己是否為隊(duì)列中的第二個(gè)節(jié)點(diǎn)
// 成為隊(duì)列中第二個(gè)節(jié)點(diǎn)才有資格獲取資源
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
// 返回等待過(guò)程中是否被中斷過(guò)
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}公平鎖和非公平鎖的主要區(qū)別為了方便對(duì)比，在這里列舉了兩種鎖的上鎖過(guò)程源碼，注意紅色標(biāo)識(shí)片段 // 公平鎖上鎖過(guò)程
final void lock() {
//調(diào)用 AQS 中 acquire方法
acquire(1);
} // 非公平鎖上鎖過(guò)程
final void lock() {
// 嘗試獲取鎖，加鎖不成功則排隊(duì)。排隊(duì)之前僅有的一次插隊(duì)機(jī)會(huì)。
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}總結(jié)1、如果第一個(gè)線程嘗試獲取資源時(shí)，此時(shí)和AQS隊(duì)列無(wú)關(guān)，線程直接持有鎖。并且不會(huì)初始化隊(duì)列，如果接下來(lái)的線程都是交替執(zhí)行，那么和AQS隊(duì)列永遠(yuǎn)無(wú)關(guān)，均為線程直接持有鎖。 2、在線程發(fā)生資源競(jìng)爭(zhēng)的情況下，才會(huì)初始化AQS隊(duì)列，AQS隊(duì)列的頭部永遠(yuǎn)是一個(gè)虛擬的Thread為NULL的node。 3、未能獲取到資源的線程將會(huì)處于park狀態(tài)，此時(shí)只有隊(duì)列中第二個(gè)node等待被喚醒，嘗試去獲取資源。其他node并不去競(jìng)爭(zhēng)資源，這也是AQS隊(duì)列的精髓所在，減少了CPU的占用。 4、公平鎖的上鎖是必須判斷自己是不是需要排隊(duì)；而非公平鎖是直接進(jìn)行CAS修改計(jì)數(shù)器看能不能加鎖成功；如果加鎖不成功則乖乖排隊(duì)(調(diào)用acquire)；所以不管公平還是不公平；只要進(jìn)到了AQS隊(duì)列當(dāng)中那么他就會(huì)排隊(duì)；一朝排隊(duì)；永遠(yuǎn)排隊(duì)！ 來(lái)源：https://www./content-4-397501.html |
|
|
來(lái)自： 印度阿三17 > 《開(kāi)發(fā)》
