AQS是什麼

java concurrent包中有很多阻塞類如：ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore、Synchronous、FutureTask等，他們的底層都是根據aqs構建的，它可以說是java多執行緒程式設計最底層核心的抽象類。既然這麼重要，我們就來看看它底層原理到底是什麼。

aqs全稱AbstractQueuedSynchronizer，它作為抽象類無法單獨使用，需要有具體實現，不同的實現中自己定義什麼狀態意味著獲取或者被釋放

AQS的原理是什麼

AQS內部維護一個先進先出（FIFO）的等待佇列叫做CLH佇列，當一個執行緒來請求資源時，AQS通過狀態判斷是否能獲取資源，如果不能獲取，則掛起這個執行緒，和狀態一起封裝成一個Node節點放在隊尾，等待前面的執行緒釋放資源好喚醒自己，所以誰先請求的誰最先獲得機會喚醒,當然新執行緒可能加塞提前獲取資源，在原始碼解析可以看到原因

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 掛起狀態

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才會走這裡，cas設定成-1待喚醒狀態
            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，這樣往復，釋放鎖後繼續喚醒後面的執行緒。