一、佇列同步器 AbstractQueuedSynchronizer

佇列同步器 AbstractQueuedSynchronizer，是用來構建鎖或者其他同步組件的基礎框架，它使用了一個int成員變數表示同步狀態，通過內置的FIFO 佇列來完成資源獲取執行緒的排隊作業，

1. Lock介面

Java SE 5 之后，并發包中新增了 Lock 介面(以及相關實作類)，用來實作鎖功能，它提供了與 synchronized 關鍵字類似的同步功能，只是在使用時需要顯式地獲取鎖和釋放鎖，雖然它缺少了隱式獲取釋放鎖的便捷性，但是卻擁有了鎖獲取與同步的可操作性，可中斷的獲取鎖以及超時獲取鎖等多種synchronized 關鍵字不具備的功能

Lock 介面提供的synchronized 關鍵字不具備的主要特性

• 嘗試非阻塞地獲取鎖：當前執行緒嘗試獲取鎖，如果這一時刻鎖沒有被其他執行緒獲取到，則成功獲取并持有鎖
• 能被中斷的獲取鎖：獲取到鎖的執行緒能夠回應中斷，當獲取到鎖的執行緒被中斷時，中斷例外將會拋出，同時鎖會被釋放
• 超時獲取鎖：在指定截止時間之前獲取鎖，如果截止時間到了仍舊無法獲取鎖，則回傳

2. AQS的使用

AQS的設計是基于模板方法模式的，也就是說，使用者需要繼承同步器并重寫指定的方法，隨后將同步器組合在自定義同步組件的實作中，并呼叫同步器提供的模板方法，而這些模板方法將會呼叫使用者重寫的方法，這樣就大大降低了實作一個可靠自定義同步組件的門檻

package lockTest;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @Description: AQS
 * @Author: Aiguodala
 * @CreateDate: 2021/4/19 10:51
 */

public class AQSTest {

    private static final MyLock LOCK = new MyLock();
    public static void main(String[] args) {
        new Thread(() -> {
            LOCK.lock();
            try {
                System.out.println("第一個執行緒獲取鎖");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("第一個執行緒釋放鎖");
                LOCK.unlock();
            }
        }).start();

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Thread(() -> {
            LOCK.lock();
            try {
                System.out.println("第二個執行緒獲取鎖");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("第二個執行緒釋放鎖");
                LOCK.unlock();
            }
        }).start();




    }
}
class MyLock implements Lock {

    /**
     * 靜態內部類，自定義同步器
     */
    private static class Sync extends AbstractQueuedSynchronizer {

        /**
         * 是否處于占用狀態
         * @return
         */
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        /**
         * 當狀態為 0 的時候獲取鎖
         * @param arg
         * @return
         */
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0,1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        /**
         * 釋放鎖，將狀態置為 0
         * @param arg
         * @return
         */
        @Override
        protected boolean tryRelease(int arg) {
            if (getState() == 0) {
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            // setState 放在最后是因為 state 變數是volatile 修飾的，加入寫屏障可以保證可見性
            setState(0);
            return true;
        }

        /**
         * 每個Condition 都包含了一個Condition佇列
         * @return
         */
        Condition newCondition () {
            return new ConditionObject();
        }
    }

    /*********** 以下方法只需要將操作代理到 Sync 物件上 ***********/

    /**
     * 創建鎖物件
     */
    private final Sync sync = new Sync();

    @Override
    public void lock() {
        sync.acquire(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1,unit.toNanos(time));
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }
}

3. AQS 實作原理

3.1 同步佇列

同步器依賴內部的同步佇列來完成同步狀態的管理，當前執行緒獲取同步狀態失敗時，同步器會將當前執行緒以及等待狀態的等資訊構造成一個節點( Node )并將其加入到同步佇列，同時會阻塞當前執行緒，當同步狀態釋放時，會把首節點中的執行緒喚醒，使其再次嘗試獲取同步狀態，

Node 主要欄位

        /**
         * 等待狀態
         */
        volatile int waitStatus;

        /**
         * 前驅節點，當節點被加入到等待佇列時被添加
         */
        volatile Node prev;

        /**
         * 后繼節點
         */
        volatile Node next;

        /**
         * 獲取同步狀態的執行緒
         */
        volatile Thread thread;

        /**
         * 等待佇列的后繼節點，如果當前節點是共享的，那么這個欄位將是一個SHARED常量，
         * 也就是說節點型別（獨占或者共享）和等待佇列中的后繼節點共用一個欄位
         */
        Node nextWaiter;

• waitStatus 的狀態有：

        /**
         * 由于在同步佇列中等待的執行緒等待超時或者被中斷，需要從同步佇列中取消等待
         * 節點變成該狀態將不會變化
         */
        static final int CANCELLED =  1;
        
        /**
         * 后繼節點的執行緒處于等待狀態，而當前節點的執行緒如果釋放了同步狀態或者被取消
         * 將會通知后繼節點，使得后繼節點得以運行
         */
        static final int SIGNAL    = -1;
        
        /**
         * 節點在等待佇列中，節點執行緒等待在Condition上
         * 當其他執行緒對Condition呼叫了signal() 方法后，該節點會從等待佇列轉移到同步佇列中
         */
        static final int CONDITION = -2;
        
        /**
         * 表示下一次共享式同步狀態會被無條件傳播下去
         */
        static final int PROPAGATE = -3;

佇列結構

• 同步器包含兩個節點型別的參考，一個指向頭結點，一個指向尾結點
• 一個執行緒成功獲取鎖后，其他執行緒無法獲得鎖而進入同步佇列，這個操作必須保證執行緒安全，利用CAS加入尾結點
• 首節點完成任務后釋放鎖并且喚醒后繼節點，因為只有一個執行緒能夠獲得鎖，所以不需要使用CAS

3.2 獨占式同步狀態獲取和釋放

① acquire( int arg)

獲取同步狀態，該方法對中斷不敏感，也就是由于執行緒獲取同步狀態失敗后進入同步佇列中，后續對執行緒進行中斷操作時，執行緒不會從同步佇列中移除

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

• 首先呼叫自己重寫的tryAcquire(arg) 方法，保證執行緒安全的獲取狀態并且修改為同步狀態
• 如果失敗就構造節點(獨占式Node.EXCLUSIVE) 通過addWaiter 加入到佇列尾部
• 最后呼叫acquireQueued () 方法使得節點以死回圈方式獲取同步住哪個臺，獲取不到則阻塞節點保存的執行緒，

addWaiter （Node node）

    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // 備份tail節點參考
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            // 利用CAS 設定尾結點
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 不成功進入enq
        enq(node);
        return node;
    }

enq(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;
                // 只有通過CAS成功添加節點才會退出
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            // 死回圈，不斷自旋，獲取同步狀態才可以退出
            for (;;) {
            	// 獲取node 的前驅節點
                final Node p = node.predecessor();
                // 只有前驅節點是頭結點，才能嘗試獲取同步狀態
                if (p == head && tryAcquire(arg)) {
                	// 將當前節點設定為頭結點
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // shouldParkAfterFailedAcquire 將前驅節點狀態置為 -1 用于標記誰來喚醒該被阻塞地節點
                // 如果 shouldParkAfterFailedAcquire 回傳true 則進入parkAndCheckInterrupt 節點被阻塞
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

② release(int arg)

當前執行緒獲取同步狀態并執行相應邏輯之后，需要釋放同步狀態，使得后續節點能夠繼續獲取同步狀態，

    public final boolean release(int arg) {
    	// 呼叫自己重寫的tryRelease(arg) 方法
        if (tryRelease(arg)) {
            Node h = head;
            // 喚醒處于等待的執行緒
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

3.3 共享式同步狀態獲取和釋放

共享式獲取與獨占式獲取最重要的區別在于同一時刻是否能有多個執行緒同時獲取到同步狀態，
例如：讀檔案是共享式，而寫檔案是獨占式

① acquireShared(int arg)

    public final void acquireShared(int arg) {
    	// 呼叫自定義的tryAcquireShared(arg) 方法，回傳值大于0 則獲取同步狀態
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }


    private void doAcquireShared(int arg) {
    	// 構造SHARED 共享式節點
        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);
        }
    }

② acquireShared(int arg)

    public final boolean releaseShared(int arg) {
    	// 呼叫自己重寫的tryReleaseShared(arg) 方法嘗試釋放同步狀態
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                	// 必須確保同步狀態的執行緒安全釋放，因為釋放同步操作的執行緒
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

3.4 獨占式超時獲取同步狀態

通過呼叫同步器的doAcquireNanos(int arg, long nanosTimeout) 方法，可以超時獲取同步狀態，提供了synchronized所不具備的特性

實作邏輯和獨占式獲取同步狀態大體相同，只是在獲取失敗時會判斷是否超時

    private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }