AQS是什么
#Java# concurrent包中有很多阻塞類如：ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore、Synchronous、FutureTask等，他們的底層都是根據aqs構建的，它可以說是JAVA多線程編程最底層核心的抽象類。既然這么重要，我們就來看看它底層原理到底是什么。
aqs全稱AbstractQueuedSynchronizer，它作為抽象類無法單獨使用，需要有具體實現，不同的實現中自己定義什么狀態意味著獲取或者被釋放
AQS的原理是什么
AQS內部維護一個先進先出（FIFO）的等待隊列叫做CLH隊列，當一個線程來請求資源時，AQS通過狀態判斷是否能獲取資源，如果不能獲取，則掛起這個線程，和狀態一起封裝成一個Node節點放在隊尾，等待前面的線程釋放資源好喚醒自己，所以誰先請求的誰最先獲得機會喚醒,當然新線程可能加塞提前獲取資源，在源碼解析可以看到原因
aqs.jpg
AQS分獨占和共享兩種方式，獨占模式，只有一個線程可以獲得鎖，比如ReentrantLock，共享模式下可以允許多個線程同時獲取鎖，比如CountDownLatch使用的就是共享方式，
源碼解析
AQS的子類需要實現的方法
//獨占方式獲取資源
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
//獨占釋放資源
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
//共享獲取資源
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
//共享釋放資源
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
//是否獨占
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
可以看到，子類調用這些方法如果沒有實現的話會拋異常，當然也不是所有方法都要實現，找自己需要的實現就可以了。
為了更好的理解先實現一個最簡單的鎖,只需要實現tryAcquire和tryRelease方法即可
public class TestLock {
private Sync sync = new Sync();
//加鎖
public void lock(){
sync.acquire(1);
}
//解鎖
public void unLock(){
sync.release(1);
}
public static class Sync extends AbstractQueuedSynchronizer {
@Override
protected boolean tryAcquire(int arg) {
assert arg == 1;
//cas將狀態從0設為1，如何不為0則失敗
if(compareAndSetState(0,1)){
return true;
}
return false;
}
@Override
protected boolean tryRelease(int arg) {
assert arg == 1;
if(getState() == 0){
throw new IllegalMonitorStateException();
}
//將狀態設為0
setState(0);
return true;
}
}
}
再來寫一個并發場景，簡單的加法，先獲取前值，用sleep模擬方法執行時間比較長,然后累加
public static void main(String[] args) {
final AddCount count = new AddCount();
ExecutorService executorService = Executors.newCachedThreadPool();
for(int i = 0;i<3;i++){
executorService.submit(new Runnable() {
@Override
public void run() {
try {
count.add(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
public static class AddCount{
private int countTotle = 0;
public void add(int count) throws InterruptedException {
int tmp = this.countTotle;
Thread.sleep(100L);
this.countTotle = tmp+count;
System.out.println(this.countTotle);
}
}
//輸出
100
100
100
在add方法加上自定義的的鎖
public static void main(String[] args) {
final AddCount count = new AddCount();
final TestLock testLock = new TestLock();
ExecutorService executorService = Executors.newCachedThreadPool();
for(int i = 0;i<3;i++){
executorService.submit(new Runnable() {
@Override
public void run() {
try {
testLock.lock();
count.add(100);
testLock.unLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
//輸出
100
200
300
根據這個簡單的例子，我們來看一下源碼中是怎么實現的
acquire
lock方法首先調用的是AQS的acquire方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
它會調用tryAcquire嘗試去取鎖，如果沒有取到的話調用addWaiter將Node放入隊尾，同樣也使用CAS的方式，AQS中有大量CAS的使用，不了解CAS的可以看淺析樂觀鎖、悲觀鎖與CAS
這里有新的線程在執行第一個判斷!tryAcquire(arg)時，如果剛好有線程釋放鎖，那新的線程很有可能插隊直接獲取到鎖，也就是有隊列也無法公平的原因。
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
在尾部添加node，將node雙向關聯，如果成功則直接返回，這里有一個問題，在設置隊尾的時候，沒有并發控制，有另一個線程也來設置，就只會有一個線程成功，沒成功的線程或者隊尾為空則執行enq方法。
enq方法
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;
}
}
}
}
這里看到如果tail是null,則cas設置head為一個新節點,也就是說第一個入隊的節點head和tail是相同的。 如果隊尾不為空，則用cas加自旋的方式放入隊尾。
Node對象
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;
//等待條件
static final int CONDITION = -2;
//對應共享類型釋放資源時，傳播喚醒線程狀態
static final int PROPAGATE = -3;
//當前狀態
volatile int waitStatus;
//前一個節點
volatile Node prev;
//下一個節點
volatile Node next;
//請求的線程
volatile Thread thread;
Node nextWaiter;
final boolean isShared() {
return nextWaiter == SHARED;
}
//獲取前一個節點，為空則拋空指針異常
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
沒有使用condition，node常用的狀態有 0 新建狀態和 -1 掛起狀態， waitStatus>0表示取消狀態，而waitStatus<0表示有效狀態。
acquireQueued
再看一下acquireQueued方法
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//獲取前一個節點
final Node p = node.predecessor();
//如果前一個節點是head,并且能獲取鎖，則將當前節點設置為head
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//判斷前面節點的狀態，中斷當前線程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
//失敗了設置成取消狀態
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//如果前一個節點已經是等待狀態，可以安全park
if (ws == Node.SIGNAL)
return true;
//如何前一個節點是取消狀態了，則一直往前取，去掉取消狀態的節點，直到狀態不為取消狀態的節點
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//ws必須是0或-3才會走這里，第一個入隊的節點是0狀態，cas設置成-1待喚醒狀態，下一次循環就返回true了，將線程掛起
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//中斷當前線程
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
這里的主要邏輯就是將新加入的節點設置為待喚醒狀態，第一次搶到鎖的線程不會進入隊列，只有后續沒有搶到的線程才進隊列，進入隊列的節點都進入中斷狀態，搶到鎖的線程釋放鎖后喚醒head節點持有鎖，鎖被釋放后會繼續喚醒后面的節點代替之前的head成為新的head節點
release
釋放鎖的過程，調用release方法
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
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);
}
調用LockSupport.unpark后，喚醒后一個中斷的線程，隊列剔除之前的head，這樣往復，釋放鎖后繼續喚醒后面的線程。
