圖解AQS的設計與實現，手摸手帶你實現一把互斥鎖！
AQS是併發編程中很是重要的概念，它是juc包下的許多併發工具類，如CountdownLatch，CyclicBarrier，Semaphore 和鎖, 如ReentrantLock， ReaderWriterLock的實現基礎，提供了一個基於int狀態碼和隊列來實現的併發框架。本文將對AQS框架的幾個重要組成進行簡要介紹，讀完本文你將get到如下幾個點：java
- AQS進行併發控制的機制是什麼
- 共享模式和獨佔模式獲取和釋放同步狀態的詳細過程
- 基於AQS框架實現一個簡易的互斥鎖
一，AQS基本概念
AQS（AbstractQueuedSynchronizer）是用來構建鎖或者其餘同步組件的基礎框架，它使用了一個int成員變量來表示狀態，經過內置的FIFO（first in，first out）隊列來完成資源獲取線程的排隊工做。node
1.1 同步狀態
AQS中維持一個全局的int狀態碼（state），線程經過修改（加/減指定的數量）碼是否成功來決定當前線程是否成功獲取到同步狀態。編程
1.1 獨佔or共享模式
AQS支持兩種獲取同步狀態的模式既獨佔式和共享式。顧名思義，獨佔式模式同一時刻只容許一個線程獲取同步狀態，而共享模式則容許多個線程同時獲取。多線程
1.2 同步隊列
同步隊列（一個FIFO雙向隊列）是AQS的核心，用來完成同步狀態的管理，當線程獲取同步狀態失敗時，AQS會將當前線程以及等待狀態等信息構形成一個節點並加入到同步隊列，同時會阻塞當前線程。併發
二，獨佔模式獲取與釋放狀態
獨佔模式既同一時間只能由一個線程持有同步狀態。當多個線程競爭時(acquire)，獲取到同步狀態的線程會將當前線程賦值給Thread exclusiveOwnerThread屬性（AQS父類中）來標記當前狀態被線程獨佔。其餘線程將被構形成Node加入到同步隊列中。當線程l框架
2.1 獲取同步狀態
/**
* 獲取同步狀態
*/
public final void acquire(int arg) {
/**
* 1. tryAcquire 嘗試獲取同步狀態;
* 2.1 addWaiter 若是嘗試獲取到同步狀態失敗，則加入到同步隊列中;
* 2.2 acquireQueued 在隊列中嘗試獲取同步狀態.
*/
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
多線程併發獲取（修改）同步狀態, 修改同步狀態成功的線程標記爲擁有同步狀態
ide/** * 嘗試獲取同步狀態【子類中實現】，由於aqs基於模板模式，僅提供基於狀態和同步隊列的實 * 現框架，具體的實現邏輯由子類決定 */ protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { // a. 嘗試修改狀態值操做執行成功 if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { // b. 修改狀態值成功，記錄當前持有同步狀態的線程信息 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; }獲取失敗的線程，加入到同步隊列的隊尾；加入到隊列中後，若是當前節點的前驅節點爲頭節點再次嘗試獲取同步狀態（下文代碼：p == head && tryAcquire(arg)）。工具
/** * 沒有獲取到同步狀態的線程加入到隊尾部 */ private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // 嘗試用最快的方式入隊，若是入隊失敗，再走完整的入隊方法 Node pred = tail; if (pred != null) { node.prev = pred; // 將當前線程設置到隊尾 if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } // 正常的入隊方法 enq(node); return node; } /** * 同步隊列中節點，嘗試獲取同步狀態 */ final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; // 自旋(死循環) for (;;) { // 只有當前節點的前驅節點是頭節點時纔會嘗試執行獲取同步狀態操做 final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node);// 注意： 此處重點， 當前節點設置爲頭節點，至關於頭節點出隊 p.next = null; // help GC failed = false; return interrupted; } // 獲取失敗後是否進入wait if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }- 若是頭節點的下一個節點嘗試獲取同步狀態失敗後，會進入等待狀態；其餘節點則繼續自旋。
// 僞代碼
final boolean acquireQueued(final Node node, int arg) {
for (;;) {
// -------獲取同步狀態失敗-------
// 獲取失敗後是否進入wait
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
}
/**
* 當獲取同步狀態失敗後是否進入park狀態
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// 前驅節點爲喚醒狀態，返回true【後面代碼暫時能夠忽略】
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
獨佔模式獲取同步狀態總結測試
2.2 釋放同步狀態
當線程執行完相應邏輯後，須要釋放同步狀態，使後繼節點有機會同步狀態（讓出資源，讓排隊的線程使用）。這時就須要調用release(int arg)方法。調用該方法後，會喚醒後繼節點。ui
- 釋放同步狀態，喚醒後繼節點
/**
* 釋放同步狀態
*/
public final boolean release(int arg) {
// 1. 嘗試釋放同步狀態
if (tryRelease(arg)) {
Node h = head;
// 釋放成功後，執行unpark，既喚醒操做（暫時可忽略waitStatus，涉及到條件隊列）
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
/**
* 嘗試釋放同步狀態，既將同步狀態減去指定的值
* 若是state = 0，表示當前線程 獲取次數 = 釋放次數，既釋放成功，此時將持有同步狀態線程標誌爲null
*/
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
// 狀態碼=0，表示釋放成功了
if (c == 0) {
free = true;
// 獨佔標誌設置爲null
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
/**
* 喚醒後繼節點操做
*/
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// 獲取後繼節點
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
// 喚醒後繼節點
if (s != null)
LockSupport.unpark(s.thread);
}
後繼節點獲取同步狀態成功，頭節點出隊。須要注意的事，出隊操做是間接的，有節點獲取到同步狀態時，會將當前節點設置爲head，而本來的head設置爲null。
/**
* 同步隊列中節點，嘗試獲取同步狀態（僞代碼）
* 獲取成功後，當前節點設置爲頭節點，頭節點設置爲null，既頭節點出隊
*/
final boolean acquireQueued(final Node node, int arg) {
try {
// 自旋(死循環)
for (;;) {
if (p == head && tryAcquire(arg)) {
// a. 操做：當前節點設置爲頭節點，當前節點的前驅節點設置爲null
setHead(node);
// b. 原始的head的next設置爲null，此時原始的head已經被移出隊列
p.next = null; // help GC
failed = false;
return interrupted;
}
}
}
}
/**
* a.當前節點設置爲頭節點，當前節點的前驅節點設置爲null
*/
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
2.3 其餘競爭狀況
當同步隊列中頭節點喚醒後繼節點時，此時可能有其餘線程嘗試獲取同步狀態。
假設獲取成功，將會被設置爲頭節點。
頭節點後續節點獲取同步狀態失敗。
三，共享模式獲取與釋放狀態
共享模式和獨佔模式最主要的區別是在支持同一時刻有多個線程同時獲取同步狀態。爲了不帶來額外的負擔，在上文中提到的同步隊列中都是用獨佔模式進行講述，其實同步隊列中的節點應該是獨佔和共享節點並存的。
接下來將針對共享模式狀態下獲取與釋放狀態的過程，圖文並茂得進行分析。
3.1 獲取同步狀態
首先至少要調用一次tryAcquireShared(arg)方法，若是返回值大於等於0表示獲取成功。
當獲取鎖失敗時，則建立一個共享類型的節點並進入一個同步隊列，而後進入隊列中進入自旋狀態（阻塞，喚醒兩種狀態來回切換，直到獲取到同步狀態爲止）
當隊列中的等待線程被喚醒之後就從新嘗試獲取鎖資源，若是成功則喚醒後面還在等待的共享節點並把該喚醒事件傳遞下去，即會依次喚醒在該節點後面的全部共享節點，不然繼續掛起等待。
當一個同享節點獲取到同步狀態，並喚醒後面等待的共享狀態的結果以下圖所示：
/**
* 共享模式獲取同步狀態；
* 1. 首先至少要調用一次tryAcquireShared(arg)方法，若是返回值大於等於0表示獲取成功，直接返回結果便可
* 2. 不然，將會加入到同步隊列中，反覆阻塞與喚醒，直到獲取同步狀態成功爲止； 獲取成功後會喚醒後面還在等待的共享節點並把該喚醒事件傳遞下去，即會依次喚醒在該節點後面的全部共享節點
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
/**
* 2. 自旋模式獲取同步狀態
*/
private void doAcquireShared(int arg) {
// 2.1 第一次獲取失敗後，會將此線程加入到同步隊列中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 若是前驅節點是頭節點，嘗試獲取同步狀態
final Node p = node.predecessor();
if (p == head) {
// r > 0表示獲取同步狀態成功，而且還有共享類型節點在同步隊列中
// r == 0 表示獲取同步狀態成功，同步隊列中沒有其餘共享模式節點
int r = tryAcquireShared(arg);
if (r >= 0) {
// ！！！！ 獲取同步狀態成功後，將當前node設置爲頭節點，並向後傳播，喚醒共享模式等待的節點
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);
}
}
/**
* 設置新的頭結點，並設置後面須要喚醒的節點
*/
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
// propagate > 0 代表後面須要喚醒的共享模式節點
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
// 若是當前節點的後繼節點是共享類型或者沒有後繼節點，則進行喚醒
// 這裏能夠理解爲除非明確指明不須要喚醒（後繼等待節點是獨佔類型），不然都要喚醒
if (s == null || s.isShared())
doReleaseShared();
}
}
/**
* 喚醒全部共享模式節點
*/
private void doReleaseShared() {
for (;;) {
// 喚醒操做由頭結點開始，注意這裏的頭節點已是上面新設置的頭結點了
// 其實就是喚醒上面新獲取到共享鎖的節點的後繼節點
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// 表示後繼節點須要被喚醒
if (ws == Node.SIGNAL) {
//這裏須要控制併發，由於入口有setHeadAndPropagate跟release兩個，避免兩次unpark
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
//執行喚醒操做
unparkSuccessor(h);
}
//若是後繼節點暫時不須要喚醒，則把當前節點狀態設置爲PROPAGATE確保之後能夠傳遞下去
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
// 若是頭結點沒有發生變化，表示設置完成，退出循環
// 若是頭結點發生變化，好比說其餘線程獲取到了鎖，爲了使本身的喚醒動做能夠傳遞，必須進行重試
if (h == head)
break;
}
}
最後，獲取到同步狀態的線程執行完畢，同步隊列中只有一個獨佔節點：
3.2 釋放同步狀態
釋放同步狀態後，同步隊列的變化過程和共享節點獲取到同步狀態後的變化過程一致，此處再也不進行贅述。
/**
* 釋放同步狀態，若是釋放成功，喚醒後面等待的節點
*
*/
public final boolean releaseShared(int arg) {
// 1. 嘗試釋放同步狀態
if (tryReleaseShared(arg)) {
// 2. 釋放成功後，喚醒後續等待共享節點
doReleaseShared();
return true;
}
return false;
}
四，基於AQS實現互斥鎖
讀到此處，大部分人應該還比較懵逼，似懂非懂。接下來筆者將經過AQS實現一個互斥鎖帶你打開AQS的正確打開姿式。
多線程環境count += 1可能會存在問題，詳情能夠看在併發編程bug的來源中介紹的三大緣由。正如大多數人都知道的，咱們一般可使用synchronized關鍵字進行同步，接下來咱們就基於AQS自定義一個互斥鎖來完成相同的功能。
4.1 代碼實現
/**
* 自定義互斥鎖
*
* @author cruder
* @time 2019/11/29 23:23
*/
public class MutexLock {
private static final Sync STATE_HOLDER = new Sync();
/**
* 經過Sync內部類來持有同步狀態， 當狀態爲1表示鎖被持有，0表示鎖處於空閒狀態
*/
private static class Sync extends AbstractQueuedSynchronizer {
/**
* 是否被獨佔， 有兩種表示方式
* 1. 能夠根據狀態，state=1表示鎖被佔用，0表示空閒
* 2. 能夠根據當前獨佔鎖的線程來判斷，即getExclusiveOwnerThread()!=null 表示被獨佔
*/
@Override
protected boolean isHeldExclusively() {
return getExclusiveOwnerThread() != null;
}
/**
* 嘗試獲取鎖，將狀態從0修改成1，操做成功則將當前線程設置爲當前獨佔鎖的線程
*/
@Override
protected boolean tryAcquire(int arg) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
/**
* 釋放鎖，將狀態修改成0
*/
@Override
protected boolean tryRelease(int arg) {
if (getState() == 0) {
throw new UnsupportedOperationException();
}
setExclusiveOwnerThread(null);
setState(0);
return true;
}
}
/**
* 下面的實現Lock接口須要重寫的方法，基本是就是調用內部內Sync的方法
*/
public void lock() {
STATE_HOLDER.acquire(1);
}
public void unlock() {
STATE_HOLDER.release(1);
}
}
4.2 鎖的測試
咱們定義一個計數器類，裏面定義了2個不一樣的計數方法，其中一個使用互斥鎖進行同步。開啓10個線程併發執行，每一個線程計數10000次，而後對比統計結果與預期的100，000是否相符。
package myLock;
import java.util.concurrent.*;
/**
* 自定義鎖測試
*
* @author liqiang
* @time 2019/11/29 12:39
*/
public class MyLockTest {
public static void main(String[] args) throws InterruptedException {
int threadNum = 10;
int countPerThread = 10000;
// 線程池建立的正確姿式
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(threadNum, threadNum, 1000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.AbortPolicy());
CountDownLatch countDownLatch = new CountDownLatch(threadNum);
Counter counter = new Counter();
Counter counterUnsafe = new Counter();
for (int i = 0; i < threadNum; i++) {
threadPool.submit(() -> {
for (int j = 0; j < countPerThread; j++) {
counter.getAndIncrement();
counterUnsafe.getAndIncrementUnSfae();
}
countDownLatch.countDown();
});
}
countDownLatch.await();
System.out.printf("%s 個線程，每一個線程累加了 %s 次，執行結果：safeCounter = %s, unsafeCounter = %s ", threadNum, countPerThread, counter.get(), counterUnsafe.get());
threadPool.shutdownNow();
}
}
class Counter {
private MutexLock mutexLock;
private volatile int count;
Counter() {
this.mutexLock = new MutexLock();
}
int get() {
return count;
}
int getAndIncrement() {
mutexLock.lock();
count++;
mutexLock.unlock();
return count;
}
int getAndIncrementUnSfae() {
count++;
return count;
}
}
結果和預期同樣，用自定義鎖實現的計數器統計沒有偏差。
五，總結
- AQS經過一個int同步狀態碼，和一個（先進先出）隊列來控制多個線程訪問資源
- 支持獨佔和共享兩種模式獲取同步狀態碼
- 當線程獲取同步狀態失敗會被加入到同步隊列中
- 當線程釋放同步狀態，會喚醒後繼節點來獲取同步狀態
- 共享模式下的節點獲取到同步狀態或者釋放同步狀態時，不只會喚醒後繼節點，還會向後傳播，喚醒全部同步節點
- 使用volatile關鍵字保證狀態碼在線程間的可見性，CAS操做保證修改狀態碼過程的原子性。
AQS的設計與實現比本文中描述的要稍微複雜一些，爲了達到快速入門的效果因此本文進行了簡化。對於沒有講到的內容，好比，對於獨佔模式下超時獲取同步狀態， 隊列中節點狀態的流轉， 條件隊列等沒有講到的內容，將會放到下一篇文章中進行介紹。
六，Q&A
Question1： 在java中一般使用synchronized來實現方法同步，AQS中經過CAS保證了修改同步狀態的一致性問題，那麼對比synchronized，cas有什麼優點不一樣與優點呢？你還知道其餘無鎖併發的策略嗎？
個人相關文章：
參考：
https://www.jianshu.com/p/1161d33fc1d0
《Java併發編程的藝術》
《Java併發編程實戰》
相關文章
- 1. 手把手帶你實現ToyReact框架
- 2. 用synchronized實現互斥鎖
- 3. zookeeper實現互斥鎖
- 4. 手把手實現Vuex（一）
- 5. 自旋鎖與互斥鎖的對比、手工實現自旋鎖
- 6. 手把手帶你實現一個mini版jquery
- 7. 手摸手，帶你實現代碼自動部署
- 8. 手勢解鎖的實現
- 9. 手把手帶你實現最簡單的插件化[二]（Activity）
- 10. 手把手帶你實現React的Modal組件
- 更多相關文章...
- • 現實生活中的 XML - XML 教程
- • Hibernate實現增刪改查 - Hibernate教程
- • ☆基於Java Instrument的Agent實現
- • Spring Cloud 微服務實戰(三) - 服務註冊與發現
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章