AQS(AbstractQueuedSynchronizer)詳解

AQS介紹

AQS是一個抽象類,主要是以繼承的方式使用,AQS自己沒有實現任何同步接口,僅僅是定義了同步狀態的獲取和釋放的方法來供自定義的同步組件的使用,AQS抽象類包含如下幾個方法: AQS定義兩種資源共享方式:Exclusive(獨佔,只有一個線程能執行,如ReentrantLock)和Share(共享,多個線程可同時執行,如Semaphore/CountDownLatch).共享模式時只用Sync Queue,獨佔模式有時只用Sync Queue,但若涉及Condition,則還有Condition Queue.在字類的tryAcquire,tryAcquireShared中實現公平與非公平的區分

不一樣的自定義同步器爭用共享資源的方式也不一樣.自定義同步器在實現時只須要實現共享資源state的獲取與釋放方式便可,至於具體線程等待隊列的維護(如獲取資源失敗入隊/喚醒出隊等),AQS已經在頂層實現好來.

* <pre>
     *      +------+  prev +-----+       +-----+
     * head |      | <---- |     | <---- |     |  tail
     *      +------+       +-----+       +-----+
     * </pre>
複製代碼

整個AQS分爲如下幾個部分:

• Node節點,用於存放獲取線程的節點,存在於Sync Queue,Condition Queue,這些節點主要的區分在於waitStatus的值
• Condition Queue整個隊列是用於獨佔模式中,只有用到Condition.awaitXX時纔會將node加到tail上,在使用Condition的前提是已經獲取Lock
• Sync Queue,獨佔共享的模式中均會使用到的存放Node的CLH Queue(主要特色是,隊列中總有一個dummy節點,後續節點獲取鎖的條件由前續節點決定,前繼節點在釋放lock時會喚醒sleep中的後續節點)
• ConditionObject,用於獨佔的模式,主要是線程釋放lock,加入Condition Queue,並進行相應的signal操做
• 獨佔的獲取lock(acquire,release),例如:ReentrantLock
• 共享的獲取lock(acquireShared,releaseShared),例如ReentrantReadWriteLock,Semaphore,CountDownLatch

內部類Node

Node節點是表明獲取lock的線程,存在於Condition Queue,Sync Queue裏面,而其主要就是nextWaiter(標記共享仍是獨佔),waitStatus標記node的狀態

static final class Node {
        /** 標示節點是不是共享節點(這樣的節點只存在於Sync Queue裏面) */
        static final Node SHARED = new Node();
        
        /** 獨佔模式 */
        static final Node EXCLUSIVE = null;

        /** CANCELLED說明節點已經取消獲取lock類(通常是因爲interrupt或timeout致使的) 
        * 不少時候是在cancelAcquire裏面進行設置這個標識
        */
        static final int CANCELLED =  1;
        
        /** SIGNAL標識當前節點的後繼節點須要喚醒(PS: 這個一般是在 獨佔模式下使用, 在共享模式下有時用 PROPAGATE) */
        static final int SIGNAL    = -1;
        
        /** 當前節點在Condition Queue裏面 */
        static final int CONDITION = -2;
        
        /**
         * 當前節點獲取到 lock 或進行 release lock 時, 共享模式的最終狀態是 PROPAGATE(PS:  *有可能共享模式的節點變成 PROPAGATE 以前就被其後繼節點搶佔 head 節點, 而從Sync *Queue中被踢出掉)
         
        /**
        *當前節點獲取到 lock 或進行 release lock 時, 共享模式的最終狀態是 PROPAGATE(PS: *有可能共享模式的節點變成 PROPAGATE 以前就被其後繼節點搶佔 head 節點, 而從Sync *Queue中被踢出掉)
        */
        static final int PROPAGATE = -3;

        /**
         * Status field, taking on only the values:
         *   SIGNAL:     The successor of this node is (or will soon be)
         *               blocked (via park), so the current node must
         *               unpark its successor when it releases or
         *               cancels. To avoid races, acquire methods must
         *               first indicate they need a signal,
         *               then retry the atomic acquire, and then,
         *               on failure, block.
         *   CANCELLED:  This node is cancelled due to timeout or interrupt.
         *               Nodes never leave this state. In particular,
         *               a thread with cancelled node never again blocks.
         *   CONDITION:  This node is currently on a condition queue.
         *               It will not be used as a sync queue node
         *               until transferred, at which time the status
         *               will be set to 0. (Use of this value here has
         *               nothing to do with the other uses of the
         *               field, but simplifies mechanics.)
         *   PROPAGATE:  A releaseShared should be propagated to other
         *               nodes. This is set (for head node only) in
         *               doReleaseShared to ensure propagation
         *               continues, even if other operations have
         *               since intervened.
         *   0:          None of the above
         *
         * The values are arranged numerically to simplify use.
         * Non-negative values mean that a node doesn't need to * signal. So, most code doesn't need to check for particular
         * values, just for sign.
         *
         * The field is initialized to 0 for normal sync nodes, and
         * CONDITION for condition nodes.  It is modified using CAS
         * (or when possible, unconditional volatile writes).
         */
        volatile int waitStatus;

        /**
         * Link to predecessor node that current node/thread relies on
         * for checking waitStatus. Assigned during enqueuing, and nulled
         * out (for sake of GC) only upon dequeuing.  Also, upon
         * cancellation of a predecessor, we short-circuit while
         * finding a non-cancelled one, which will always exist
         * because the head node is never cancelled: A node becomes
         * head only as a result of successful acquire. A
         * cancelled thread never succeeds in acquiring, and a thread only
         * cancels itself, not any other node.
         */
        volatile Node prev;

        /**
         * Link to the successor node that the current node/thread
         * unparks upon release. Assigned during enqueuing, adjusted
         * when bypassing cancelled predecessors, and nulled out (for
         * sake of GC) when dequeued.  The enq operation does not
         * assign next field of a predecessor until after attachment,
         * so seeing a null next field does not necessarily mean that
         * node is at end of queue. However, if a next field appears
         * to be null, we can scan prev's from the tail to * double-check. The next field of cancelled nodes is set to * point to the node itself instead of null, to make life * easier for isOnSyncQueue. */ volatile Node next; /** * The thread that enqueued this node. Initialized on * construction and nulled out after use. */ volatile Thread thread; /** * Link to next node waiting on condition, or the special * value SHARED. Because condition queues are accessed only * when holding in exclusive mode, we just need a simple * linked queue to hold nodes while they are waiting on * conditions. They are then transferred to the queue to * re-acquire. And because conditions can only be exclusive, * we save a field by using special value to indicate shared * mode. */ Node nextWaiter; /** * Returns true if node is waiting in shared mode. */ final boolean isShared() { return nextWaiter == SHARED; } /** * Returns previous node, or throws NullPointerException if null. * Use when predecessor cannot be null. The null check could * be elided, but is present to help the VM. * * @return the predecessor of this node */ final Node predecessor() throws NullPointerException { Node p = prev; if (p == null) throw new NullPointerException(); else return p; } Node() { // Used to establish initial head or SHARED marker } 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; } } 複製代碼

* <h3>Usage Examples</h3>
 *
 * <p>Here is a non-reentrant mutual exclusion lock class that uses
 * the value zero to represent the unlocked state, and one to
 * represent the locked state. While a non-reentrant lock
 * does not strictly require recording of the current owner
 * thread, this class does so anyway to make usage easier to monitor.
 * It also supports conditions and exposes
 * one of the instrumentation methods:
 *
 *  <pre> {@code
 * class Mutex implements Lock, java.io.Serializable {
 *
 *   // Our internal helper class
 *   private static class Sync extends AbstractQueuedSynchronizer {
 *     // Reports whether in locked state
 *     protected boolean isHeldExclusively() {
 *       return getState() == 1;
 *     }
 *
 *     // Acquires the lock if state is zero
 *     public boolean tryAcquire(int acquires) {
 *       assert acquires == 1; // Otherwise unused
 *       if (compareAndSetState(0, 1)) {
 *         setExclusiveOwnerThread(Thread.currentThread());
 *         return true;
 *       }
 *       return false;
 *     }
 *
 *     // Releases the lock by setting state to zero
 *     protected boolean tryRelease(int releases) {
 *       assert releases == 1; // Otherwise unused
 *       if (getState() == 0) throw new IllegalMonitorStateException();
 *       setExclusiveOwnerThread(null);
 *       setState(0);
 *       return true;
 *     }
 *
 *     // Provides a Condition
 *     Condition newCondition() { return new ConditionObject(); }
 *
 *     // Deserializes properly
 *     private void readObject(ObjectInputStream s)
 *         throws IOException, ClassNotFoundException {
 *       s.defaultReadObject();
 *       setState(0); // reset to unlocked state
 *     }
 *   }
 *
 *   // The sync object does all the hard work. We just forward to it.
 *   private final Sync sync = new Sync();
 *
 *   public void lock()                { sync.acquire(1); }
 *   public boolean tryLock()          { return sync.tryAcquire(1); }
 *   public void unlock()              { sync.release(1); }
 *   public Condition newCondition()   { return sync.newCondition(); }
 *   public boolean isLocked()         { return sync.isHeldExclusively(); }
 *   public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
 *   public void lockInterruptibly() throws InterruptedException {
 *     sync.acquireInterruptibly(1);
 *   }
 *   public boolean tryLock(long timeout, TimeUnit unit)
 *       throws InterruptedException {
 *     return sync.tryAcquireNanos(1, unit.toNanos(timeout));
 *   }
 * }}</pre>
 *
複製代碼