【JUC】JDK1.8源碼分析之AbstractQueuedSynchronizer（二）

1、前言

　　在鎖框架中，AbstractQueuedSynchronizer抽象類能夠絕不誇張的說，佔據着核心地位，它提供了一個基於FIFO隊列，能夠用於構建鎖或者其餘相關同步裝置的基礎框架。因此頗有必要好好分析。

2、AbstractQueuedSynchronizer數據結構

　　分析類，首先就要分析底層採用了何種數據結構，抓住核心點進行分析，通過分析可知，AbstractQueuedSynchronizer類的數據結構以下

　　

　　說明：AbstractQueuedSynchronizer類底層的數據結構是使用雙向鏈表，是隊列的一種實現，故也可當作是隊列，其中Sync queue，即同步隊列，是雙向鏈表，包括head結點和tail結點，head結點主要用做後續的調度。而Condition queue不是必須的，其是一個單向鏈表，只有當使用Condition時，纔會存在此單向鏈表。而且可能會有多個Condition queue。

3、AbstractQueuedSynchronizer源碼分析

　　3.1 類的繼承關係　　

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable

　　說明：從類繼承關係可知，AbstractQueuedSynchronizer繼承自AbstractOwnableSynchronizer抽象類，而且實現了Serializable接口，能夠進行序列化。而AbstractOwnableSynchronizer抽象類的源碼以下　

public abstract class AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    // 版本序列號
    private static final long serialVersionUID = 3737899427754241961L;
    // 構造函數
    protected AbstractOwnableSynchronizer() { }
    // 獨佔模式下的線程
    private transient Thread exclusiveOwnerThread;
    
    // 設置獨佔線程 
    protected final void setExclusiveOwnerThread(Thread thread) {
        exclusiveOwnerThread = thread;
    }
    
    // 獲取獨佔線程 
    protected final Thread getExclusiveOwnerThread() {
        return exclusiveOwnerThread;
    }
}

View Code

　　說明：AbstractOwnableSynchronizer抽象類中，能夠設置獨佔資源線程和獲取獨佔資源線程。分別爲setExclusiveOwnerThread與getExclusiveOwnerThread方法，這兩個方法會被子類調用。

　　3.2 類的內部類

　　AbstractQueuedSynchronizer類有兩個內部類，分別爲Node類與ConditionObject類。下面分別作介紹。

　　1. Node類　　　

static final class Node {
        // 模式，分爲共享與獨佔
        // 共享模式
        static final Node SHARED = new Node();
        // 獨佔模式
        static final Node EXCLUSIVE = null;        
        // 結點狀態
        // CANCELLED，值爲1，表示當前的線程被取消
        // SIGNAL，值爲-1，表示當前節點的後繼節點包含的線程須要運行，也就是unpark
        // CONDITION，值爲-2，表示當前節點在等待condition，也就是在condition隊列中
        // PROPAGATE，值爲-3，表示當前場景下後續的acquireShared可以得以執行
        // 值爲0，表示當前節點在sync隊列中，等待着獲取鎖
        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() {    // 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;
        }
    }

View Code

　　說明：每一個線程被阻塞的線程都會被封裝成一個Node結點，放入隊列。每一個節點包含了一個Thread類型的引用，而且每一個節點都存在一個狀態，具體狀態以下。

　　① CANCELLED，值爲1，表示當前的線程被取消。

　　② SIGNAL，值爲-1，表示當前節點的後繼節點包含的線程須要運行，須要進行unpark操做。

　　③ CONDITION，值爲-2，表示當前節點在等待condition，也就是在condition queue中。

　　④ PROPAGATE，值爲-3，表示當前場景下後續的acquireShared可以得以執行。

　　⑤ 值爲0，表示當前節點在sync queue中，等待着獲取鎖。

　　2. ConditionObject類　　

// 內部類
    public class ConditionObject implements Condition, java.io.Serializable {
        // 版本號
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        // condition隊列的頭結點
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        // condition隊列的尾結點
        private transient Node lastWaiter;

        /**
         * Creates a new {@code ConditionObject} instance.
         */
        // 構造函數
        public ConditionObject() { }

        // Internal methods

        /**
         * Adds a new waiter to wait queue.
         * @return its new wait node
         */
        // 添加新的waiter到wait隊列
        private Node addConditionWaiter() {
            // 保存尾結點
            Node t = lastWaiter;
            // If lastWaiter is cancelled, clean out.
            if (t != null && t.waitStatus != Node.CONDITION) { // 尾結點不爲空，而且尾結點的狀態不爲CONDITION
                // 清除狀態爲CONDITION的結點
                unlinkCancelledWaiters(); 
                // 將最後一個結點從新賦值給t
                t = lastWaiter;
            }
            // 新建一個結點
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null) // 尾結點爲空
                // 設置condition隊列的頭結點
                firstWaiter = node;
            else // 尾結點不爲空
                // 設置爲節點的nextWaiter域爲node結點
                t.nextWaiter = node;
            // 更新condition隊列的尾結點
            lastWaiter = node;
            return node;
        }

        /**
         * Removes and transfers nodes until hit non-cancelled one or
         * null. Split out from signal in part to encourage compilers
         * to inline the case of no waiters.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignal(Node first) {
            // 循環
            do {
                if ( (firstWaiter = first.nextWaiter) == null) // 該節點的nextWaiter爲空
                    // 設置尾結點爲空
                    lastWaiter = null;
                // 設置first結點的nextWaiter域
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null); // 將結點從condition隊列轉移到sync隊列失敗而且condition隊列中的頭結點不爲空，一直循環
        }

        /**
         * Removes and transfers all nodes.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignalAll(Node first) {
            // condition隊列的頭結點尾結點都設置爲空
            lastWaiter = firstWaiter = null;
            // 循環
            do {
                // 獲取first結點的nextWaiter域結點
                Node next = first.nextWaiter;
                // 設置first結點的nextWaiter域爲空
                first.nextWaiter = null;
                // 將first結點從condition隊列轉移到sync隊列
                transferForSignal(first);
                // 從新設置first
                first = next;
            } while (first != null);
        }

        /**
         * Unlinks cancelled waiter nodes from condition queue.
         * Called only while holding lock. This is called when
         * cancellation occurred during condition wait, and upon
         * insertion of a new waiter when lastWaiter is seen to have
         * been cancelled. This method is needed to avoid garbage
         * retention in the absence of signals. So even though it may
         * require a full traversal, it comes into play only when
         * timeouts or cancellations occur in the absence of
         * signals. It traverses all nodes rather than stopping at a
         * particular target to unlink all pointers to garbage nodes
         * without requiring many re-traversals during cancellation
         * storms.
         */
        // 從condition隊列中清除狀態爲CANCEL的結點
        private void unlinkCancelledWaiters() {
            // 保存condition隊列頭結點
            Node t = firstWaiter;
            Node trail = null;
            while (t != null) { // t不爲空
                // 下一個結點
                Node next = t.nextWaiter;
                if (t.waitStatus != Node.CONDITION) { // t結點的狀態不爲CONDTION狀態
                    // 設置t節點的額nextWaiter域爲空
                    t.nextWaiter = null;
                    if (trail == null) // trail爲空
                        // 從新設置condition隊列的頭結點
                        firstWaiter = next;
                    else // trail不爲空
                        // 設置trail結點的nextWaiter域爲next結點
                        trail.nextWaiter = next;
                    if (next == null) // next結點爲空
                        // 設置condition隊列的尾結點
                        lastWaiter = trail;
                }
                else // t結點的狀態爲CONDTION狀態
                    // 設置trail結點
                    trail = t;
                // 設置t結點
                t = next;
            }
        }

        // public methods

        /**
         * Moves the longest-waiting thread, if one exists, from the
         * wait queue for this condition to the wait queue for the
         * owning lock.
         *
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        // 喚醒一個等待線程。若是全部的線程都在等待此條件，則選擇其中的一個喚醒。在從 await 返回以前，該線程必須從新獲取鎖。
        public final void signal() {
            if (!isHeldExclusively()) // 不被當前線程獨佔，拋出異常
                throw new IllegalMonitorStateException();
            // 保存condition隊列頭結點
            Node first = firstWaiter;
            if (first != null) // 頭結點不爲空
                // 喚醒一個等待線程
                doSignal(first);
        }

        /**
         * Moves all threads from the wait queue for this condition to
         * the wait queue for the owning lock.
         *
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        // 喚醒全部等待線程。若是全部的線程都在等待此條件，則喚醒全部線程。在從 await 返回以前，每一個線程都必須從新獲取鎖。
        public final void signalAll() {
            if (!isHeldExclusively()) // 不被當前線程獨佔，拋出異常
                throw new IllegalMonitorStateException();
            // 保存condition隊列頭結點
            Node first = firstWaiter;
            if (first != null) // 頭結點不爲空
                // 喚醒全部等待線程
                doSignalAll(first);
        }

        /**
         * Implements uninterruptible condition wait.
         * <ol>
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * </ol>
         */
        // 等待，當前線程在接到信號以前一直處於等待狀態，不響應中斷
        public final void awaitUninterruptibly() {
            // 添加一個結點到等待隊列
            Node node = addConditionWaiter();
            // 獲取釋放的狀態
            int savedState = fullyRelease(node);
            boolean interrupted = false;
            while (!isOnSyncQueue(node)) { // 
                // 阻塞當前線程
                LockSupport.park(this);
                if (Thread.interrupted()) // 當前線程被中斷
                    // 設置interrupted狀態
                    interrupted = true; 
            }
            if (acquireQueued(node, savedState) || interrupted) // 
                selfInterrupt();
        }

        /*
         * For interruptible waits, we need to track whether to throw
         * InterruptedException, if interrupted while blocked on
         * condition, versus reinterrupt current thread, if
         * interrupted while blocked waiting to re-acquire.
         */

        /** Mode meaning to reinterrupt on exit from wait */
        private static final int REINTERRUPT =  1;
        /** Mode meaning to throw InterruptedException on exit from wait */
        private static final int THROW_IE    = -1;

        /**
         * Checks for interrupt, returning THROW_IE if interrupted
         * before signalled, REINTERRUPT if after signalled, or
         * 0 if not interrupted.
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0; 
        }

        /**
         * Throws InterruptedException, reinterrupts current thread, or
         * does nothing, depending on mode.
         */
        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

        /**
         * Implements interruptible condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled or interrupted.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        // // 等待，當前線程在接到信號或被中斷以前一直處於等待狀態
        public final void await() throws InterruptedException {
            if (Thread.interrupted()) // 當前線程被中斷，拋出異常
                throw new InterruptedException();
            // 在wait隊列上添加一個結點
            Node node = addConditionWaiter();
            // 
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                // 阻塞當前線程
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) // 檢查結點等待時的中斷類型
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

        /**
         * Implements timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        // 等待，當前線程在接到信號、被中斷或到達指定等待時間以前一直處於等待狀態 
        public final long awaitNanos(long nanosTimeout)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }

        /**
         * Implements absolute timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * <li> If timed out while blocked in step 4, return false, else true.
         * </ol>
         */
        // 等待，當前線程在接到信號、被中斷或到達指定最後期限以前一直處於等待狀態
        public final boolean awaitUntil(Date deadline)
                throws InterruptedException {
            long abstime = deadline.getTime();
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (System.currentTimeMillis() > abstime) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                LockSupport.parkUntil(this, abstime);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        /**
         * Implements timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * <li> If timed out while blocked in step 4, return false, else true.
         * </ol>
         */
        // 等待，當前線程在接到信號、被中斷或到達指定等待時間以前一直處於等待狀態。此方法在行爲上等效於：awaitNanos(unit.toNanos(time)) > 0
        public final boolean await(long time, TimeUnit unit)
                throws InterruptedException {
            long nanosTimeout = unit.toNanos(time);
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        //  support for instrumentation

        /**
         * Returns true if this condition was created by the given
         * synchronization object.
         *
         * @return {@code true} if owned
         */
        final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
            return sync == AbstractQueuedSynchronizer.this;
        }

        /**
         * Queries whether any threads are waiting on this condition.
         * Implements {@link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}.
         *
         * @return {@code true} if there are any waiting threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        //  查詢是否有正在等待此條件的任何線程
        protected final boolean hasWaiters() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    return true;
            }
            return false;
        }

        /**
         * Returns an estimate of the number of threads waiting on
         * this condition.
         * Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}.
         *
         * @return the estimated number of waiting threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        // 返回正在等待此條件的線程數估計值
        protected final int getWaitQueueLength() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            int n = 0;
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION)
                    ++n;
            }
            return n;
        }

        /**
         * Returns a collection containing those threads that may be
         * waiting on this Condition.
         * Implements {@link AbstractQueuedSynchronizer#getWaitingThreads(ConditionObject)}.
         *
         * @return the collection of threads
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        // 返回包含那些可能正在等待此條件的線程集合
        protected final Collection<Thread> getWaitingThreads() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            ArrayList<Thread> list = new ArrayList<Thread>();
            for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
                if (w.waitStatus == Node.CONDITION) {
                    Thread t = w.thread;
                    if (t != null)
                        list.add(t);
                }
            }
            return list;
        }
    }

View Code

　　說明：此類實現了Condition接口，Condition接口定義了條件操做規範，具體以下　

public interface Condition {

    // 等待，當前線程在接到信號或被中斷以前一直處於等待狀態
    void await() throws InterruptedException;
    
    // 等待，當前線程在接到信號以前一直處於等待狀態，不響應中斷
    void awaitUninterruptibly();
    
    //等待，當前線程在接到信號、被中斷或到達指定等待時間以前一直處於等待狀態 
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    
    // 等待，當前線程在接到信號、被中斷或到達指定等待時間以前一直處於等待狀態。此方法在行爲上等效於：awaitNanos(unit.toNanos(time)) > 0
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    
    // 等待，當前線程在接到信號、被中斷或到達指定最後期限以前一直處於等待狀態
    boolean awaitUntil(Date deadline) throws InterruptedException;
    
    // 喚醒一個等待線程。若是全部的線程都在等待此條件，則選擇其中的一個喚醒。在從 await 返回以前，該線程必須從新獲取鎖。
    void signal();
    
    // 喚醒全部等待線程。若是全部的線程都在等待此條件，則喚醒全部線程。在從 await 返回以前，每一個線程都必須從新獲取鎖。
    void signalAll();
}

View Code

　　說明：Condition接口中定義了await、signal函數，用來等待條件、釋放條件。以後會詳細分析CondtionObject的源碼。

　　3.3 類的屬性　　

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {    
    // 版本號
    private static final long serialVersionUID = 7373984972572414691L;    
    // 頭結點
    private transient volatile Node head;    
    // 尾結點
    private transient volatile Node tail;    
    // 狀態
    private volatile int state;    
    // 自旋時間
    static final long spinForTimeoutThreshold = 1000L;
    
    // Unsafe類實例
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    // state內存偏移地址
    private static final long stateOffset;
    // head內存偏移地址
    private static final long headOffset;
    // state內存偏移地址
    private static final long tailOffset;
    // tail內存偏移地址
    private static final long waitStatusOffset;
    // next內存偏移地址
    private static final long nextOffset;
    // 靜態初始化塊
    static {
        try {
            stateOffset = unsafe.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
            headOffset = unsafe.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
            tailOffset = unsafe.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
            waitStatusOffset = unsafe.objectFieldOffset
                (Node.class.getDeclaredField("waitStatus"));
            nextOffset = unsafe.objectFieldOffset
                (Node.class.getDeclaredField("next"));

        } catch (Exception ex) { throw new Error(ex); }
    }
}

View Code

　　說明：屬性中包含了頭結點head，尾結點tail，狀態state、自旋時間spinForTimeoutThreshold，還有AbstractQueuedSynchronizer抽象的屬性在內存中的偏移地址，經過該偏移地址，能夠獲取和設置該屬性的值，同時還包括一個靜態初始化塊，用於加載內存偏移地址。

　　3.4 類的構造函數　　

protected AbstractQueuedSynchronizer() { }    

　　說明：此類構造函數爲從抽象構造函數，供子類調用。

　　3.5 類的核心函數

　　1. acquire函數

　　該函數以獨佔模式獲取(資源)，忽略中斷，即線程在aquire過程當中，中斷此線程是無效的。源碼以下　

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

　　由上述源碼能夠知道，當一個線程調用acquire時，調用方法流程以下。

　　

　　說明：

　　① 首先調用tryAcquire函數，調用此方法的線程會試圖在獨佔模式下獲取對象狀態。此方法應該查詢是否容許它在獨佔模式下獲取對象狀態，若是容許，則獲取它。在AbstractQueuedSynchronizer源碼中默認會拋出一個異常，即須要子類去重寫此函數完成本身的邏輯。以後會進行分析。

　　② 若tryAcquire失敗，則調用addWaiter函數，addWaiter函數完成的功能是將調用此方法的線程封裝成爲一個結點並放入Sync queue。

　　③ 調用acquireQueued函數，此函數完成的功能是Sync queue中的結點不斷嘗試獲取資源，若成功，則返回true，不然，返回false。

　　因爲tryAcquire默認實現是拋出異常，因此此時，不進行分析，以後會結合一個例子進行分析。

　　首先分析addWaiter函數

// 添加等待者
    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域鏈接到尾結點
            node.prev = pred; 
            if (compareAndSetTail(pred, node)) { // 比較pred是否爲尾結點，是則將尾結點設置爲node 
                // 設置尾結點的next域爲node
                pred.next = node;
                return node; // 返回新生成的結點
            }
        }
        enq(node); // 尾結點爲空(即尚未被初始化過)，或者是compareAndSetTail操做失敗，則入隊列
        return node;
    }

　　說明：addWaiter函數使用快速添加的方式往sync queue尾部添加結點，若是sync queue隊列尚未初始化，則會使用enq插入隊列中，enq方法源碼以下　

// 入隊列
    private Node enq(final Node node) {
        for (;;) { // 無限循環，確保結點可以成功入隊列
            // 保存尾結點
            Node t = tail;
            if (t == null) { // 尾結點爲空，即還沒被初始化
                if (compareAndSetHead(new Node())) // 頭結點爲空，並設置頭結點爲新生成的結點
                    tail = head; // 頭結點與尾結點都指向同一個新生結點
            } else { // 尾結點不爲空，即已經被初始化過
                // 將node結點的prev域鏈接到尾結點
                node.prev = t; 
                if (compareAndSetTail(t, node)) { // 比較結點t是否爲尾結點，如果則將尾結點設置爲node
                    // 設置尾結點的next域爲node
                    t.next = node; 
                    return t; // 返回尾結點
                }
            }
        }
    }

　　說明：enq函數會使用無限循環來確保節點的成功插入。

　　如今，分析acquireQueue函數。其源碼以下

// sync隊列中的結點在獨佔且忽略中斷的模式下獲取(資源)
    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; 
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

 　　說明：首先獲取當前節點的前驅節點，若是前驅節點是頭結點而且可以獲取(資源)，表明該當前節點可以佔有鎖，設置頭結點爲當前節點，返回。不然，調用shouldParkAfterFailedAcquire和parkAndCheckInterrupt函數，首先，咱們看shouldParkAfterFailedAcquire函數，代碼以下

// 當獲取(資源)失敗後，檢查而且更新結點狀態
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 獲取前驅結點的狀態
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL) // 狀態爲SIGNAL，爲-1
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            // 能夠進行park操做
            return true; 
        if (ws > 0) { // 表示狀態爲CANCELLED，爲1
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0); // 找到pred結點前面最近的一個狀態不爲CANCELLED的結點
            // 賦值pred結點的next域
            pred.next = node; 
        } else { // 爲PROPAGATE -3 或者是0 表示無狀態,(爲CONDITION -2時，表示此節點在condition queue中) 
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            // 比較並設置前驅結點的狀態爲SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL); 
        }
        // 不能進行park操做
        return false;
    }

　　說明：只有當該節點的前驅結點的狀態爲SIGNAL時，才能夠對該結點所封裝的線程進行park操做。不然，將不能進行park操做。再看parkAndCheckInterrupt函數，源碼以下

// 進行park操做而且返回該線程是否被中斷
    private final boolean parkAndCheckInterrupt() {
        // 在許可可用以前禁用當前線程，而且設置了blocker
        LockSupport.park(this);
        return Thread.interrupted(); // 當前線程是否已被中斷，並清除中斷標記位
    }

　　說明：parkAndCheckInterrupt函數裏的邏輯是首先執行park操做，即禁用當前線程，而後返回該線程是否已經被中斷。再看final塊中的cancelAcquire函數，其源碼以下

// 取消繼續獲取(資源)
    private void cancelAcquire(Node node) {
        // Ignore if node doesn't exist
        // node爲空，返回
        if (node == null)
            return;
        // 設置node結點的thread爲空
        node.thread = null;

        // Skip cancelled predecessors
        // 保存node的前驅結點
        Node pred = node.prev;
        while (pred.waitStatus > 0) // 找到node前驅結點中第一個狀態小於0的結點，即不爲CANCELLED狀態的結點
            node.prev = pred = pred.prev;

        // predNext is the apparent node to unsplice. CASes below will
        // fail if not, in which case, we lost race vs another cancel
        // or signal, so no further action is necessary.
        // 獲取pred結點的下一個結點
        Node predNext = pred.next;

        // Can use unconditional write instead of CAS here.
        // After this atomic step, other Nodes can skip past us.
        // Before, we are free of interference from other threads.
        // 設置node結點的狀態爲CANCELLED
        node.waitStatus = Node.CANCELLED;

        // If we are the tail, remove ourselves.
        if (node == tail && compareAndSetTail(node, pred)) { // node結點爲尾結點，則設置尾結點爲pred結點
            // 比較並設置pred結點的next節點爲null
            compareAndSetNext(pred, predNext, null); 
        } else { // node結點不爲尾結點，或者比較設置不成功
            // If successor needs signal, try to set pred's next-link
            // so it will get one. Otherwise wake it up to propagate.
            int ws;
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) { // （pred結點不爲頭結點，而且pred結點的狀態爲SIGNAL）或者 
                                    // pred結點狀態小於等於0，而且比較並設置等待狀態爲SIGNAL成功，而且pred結點所封裝的線程不爲空
                // 保存結點的後繼
                Node next = node.next;
                if (next != null && next.waitStatus <= 0) // 後繼不爲空而且後繼的狀態小於等於0
                    compareAndSetNext(pred, predNext, next); // 比較並設置pred.next = next;
            } else {
                unparkSuccessor(node); // 釋放node的前一個結點
            }

            node.next = node; // help GC
        }
    }

　　說明：該函數完成的功能就是取消當前線程對資源的獲取，即設置該結點的狀態爲CANCELLED，接着咱們再看unparkSuccessor函數，源碼以下

// 釋放後繼結點
    private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        // 獲取node結點的等待狀態
        int ws = node.waitStatus;
        if (ws < 0) // 狀態值小於0，爲SIGNAL -1 或 CONDITION -2 或 PROPAGATE -3
            // 比較而且設置結點等待狀態，設置爲0
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        // 獲取node節點的下一個結點
        Node s = node.next;
        if (s == null || s.waitStatus > 0) { // 下一個結點爲空或者下一個節點的等待狀態大於0，即爲CANCELLED
            // s賦值爲空
            s = null; 
            // 從尾結點開始從後往前開始遍歷
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0) // 找到等待狀態小於等於0的結點，找到最前的狀態小於等於0的結點
                    // 保存結點
                    s = t;
        }
        if (s != null) // 該結點不爲爲空，釋放許可
            LockSupport.unpark(s.thread);
    }

　　說明：該函數的做用就是爲了釋放node節點的後繼結點。

　　對於cancelAcquire與unparkSuccessor函數，以下示意圖能夠清晰的表示。

　　

　　說明：其中node爲參數，在執行完cancelAcquire函數後的效果就是unpark了s結點所包含的t4線程。

　　如今，再來看acquireQueued函數的整個的邏輯。邏輯以下

　　① 判斷結點的前驅是否爲head而且是否成功獲取(資源)。

　　② 若步驟①均知足，則設置結點爲head，以後會判斷是否finally模塊，而後返回。

　　③ 若步驟①不知足，則判斷是否須要park當前線程，是否須要park當前線程的邏輯是判斷結點的前驅結點的狀態是否爲SIGNAL，如果，則park當前結點，不然，不進行park操做。

　　④ 若park了當前線程，以後某個線程對本線程unpark後，而且本線程也得到機會運行。那麼，將會繼續進行步驟①的判斷。

　　2. release

　　以獨佔模式釋放對象，其源碼以下

public final boolean release(int arg) {
        if (tryRelease(arg)) { // 釋放成功
            // 保存頭結點
            Node h = head; 
            if (h != null && h.waitStatus != 0) // 頭結點不爲空而且頭結點狀態不爲0
                unparkSuccessor(h); //釋放頭結點的後繼結點
            return true;
        }
        return false;
    }

　　說明：其中，tryRelease的默認實現是拋出異常，須要具體的子類實現，若是tryRelease成功，那麼若是頭結點不爲空而且頭結點的狀態不爲0，則釋放頭結點的後繼結點，unparkSuccessor函數已經分析過，再也不累贅。

　　對於其餘函數咱們也能夠分析，與前面分析的函數大同小異，因此，再也不累贅。

4、示例分析

　　1. 示例一

　　藉助下面示例來分析AbstractQueuedSyncrhonizer內部的工做機制。示例源碼以下　　

package com.hust.grid.leesf.abstractqueuedsynchronizer;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class MyThread extends Thread {
    private Lock lock;
    public MyThread(String name, Lock lock) {
        super(name);
        this.lock = lock;
    }
    
    public void run () {
        lock.lock();
        try {
            System.out.println(Thread.currentThread() + " running");
        } finally {
            lock.unlock();
        }
    }
}
public class AbstractQueuedSynchonizerDemo {
    public static void main(String[] args) {
        Lock lock = new ReentrantLock();
        
        MyThread t1 = new MyThread("t1", lock);
        MyThread t2 = new MyThread("t2", lock);
        t1.start();
        t2.start();    
    }
}

　　運行結果（可能的一種）：

Thread[t1,5,main] running
Thread[t2,5,main] running

　　結果分析：從示例可知，線程t1與t2共用了一把鎖，即同一個lock。可能會存在以下一種時序。

　　說明：首先線程t1先執行lock.lock操做，而後t2執行lock.lock操做，而後t1執行lock.unlock操做，最後t2執行lock.unlock操做。基於這樣的時序，分析AbstractQueuedSynchronizer內部的工做機制。

　　① t1線程調用lock.lock函數，其函數調用順序以下，只給出了主要的函數調用。

　　說明：其中，前面的部分表示哪一個類，後面是具體的類中的哪一個方法，AQS表示AbstractQueuedSynchronizer類，AOS表示AbstractOwnableSynchronizer類。

　　② t2線程調用lock.lock函數，其函數調用順序以下，只給出了主要的函數調用。

　　說明：通過一系列的函數調用，最後達到的狀態是禁用t2線程，由於調用了LockSupport.lock。

　　③ t1線程調用lock.unlock，其函數調用順序以下，只給出了主要的函數調用。

　　說明：t1線程中調用lock.unlock後，通過一系列的調用，最終的狀態是釋放了許可，由於調用了LockSupport.unpark。這時，t2線程就能夠繼續運行了。此時，會繼續恢復t2線程運行環境，繼續執行LockSupport.park後面的語句，即進一步調用以下。

　　說明：在上一步調用了LockSupport.unpark後，t2線程恢復運行，則運行parkAndCheckInterrupt，以後，繼續運行acquireQueued函數，最後達到的狀態是頭結點head與尾結點tail均指向了t2線程所在的結點，而且以前的頭結點已經從sync隊列中斷開了。

　　④ t2線程調用lock.unlock，其函數調用順序以下，只給出了主要的函數調用。

　　說明：t2線程執行lock.unlock後，最終達到的狀態仍是與以前的狀態同樣。

　　2. 示例二

　　下面咱們結合Condition實現生產者與消費者，來進一步分析AbstractQueuedSynchronizer的內部工做機制。

　　Depot（倉庫）類

package com.hust.grid.leesf.reentrantLock;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Depot {
    private int size;
    private int capacity;
    private Lock lock;
    private Condition fullCondition;
    private Condition emptyCondition;
    
    public Depot(int capacity) {
        this.capacity = capacity;    
        lock = new ReentrantLock();
        fullCondition = lock.newCondition();
        emptyCondition = lock.newCondition();
    }
    
    public void produce(int no) {
        lock.lock();
        int left = no;
        try {
            while (left > 0) {
                while (size >= capacity)  {
                    System.out.println(Thread.currentThread() + " before await");
                    fullCondition.await();
                    System.out.println(Thread.currentThread() + " after await");
                }
                int inc = (left + size) > capacity ? (capacity - size) : left;
                left -= inc;
                size += inc;
                System.out.println("produce = " + inc + ", size = " + size);
                emptyCondition.signal();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    
    public void consume(int no) {
        lock.lock();
        int left = no;
        try {            
            while (left > 0) {
                while (size <= 0) {
                    System.out.println(Thread.currentThread() + " before await");
                    emptyCondition.await();
                    System.out.println(Thread.currentThread() + " after await");
                }
                int dec = (size - left) > 0 ? left : size;
                left -= dec;
                size -= dec;
                System.out.println("consume = " + dec + ", size = " + size);
                fullCondition.signal();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

　　測試類

package com.hust.grid.leesf.reentrantLock;


class Consumer {
    private Depot depot;
    public Consumer(Depot depot) {
        this.depot = depot;
    }
    
    public void consume(int no) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                depot.consume(no);
            }
        }, no + " consume thread").start();
    }
}

class Producer {
    private Depot depot;
    public Producer(Depot depot) {
        this.depot = depot;
    }
    
    public void produce(int no) {
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                depot.produce(no);
            }
        }, no + " produce thread").start();
    }
}

public class ReentrantLockDemo {
    public static void main(String[] args) throws InterruptedException {
        Depot depot = new Depot(500);
        new Producer(depot).produce(500);
        new Producer(depot).produce(200);
        new Consumer(depot).consume(500);
        new Consumer(depot).consume(200);
    }
}

　　運行結果（可能的一種）：　　

produce = 500, size = 500
Thread[200 produce thread,5,main] before await
consume = 500, size = 0
Thread[200 consume thread,5,main] before await
Thread[200 produce thread,5,main] after await
produce = 200, size = 200
Thread[200 consume thread,5,main] after await
consume = 200, size = 0

　　說明：根據結果，咱們猜想一種可能的時序以下

　　說明：p1表明produce 500的那個線程，p2表明produce 200的那個線程，c1表明consume 500的那個線程，c2表明consume 200的那個線程。

　　1. p1線程調用lock.lock，得到鎖，繼續運行，函數調用順序在前面已經給出。

　　2. p2線程調用lock.lock，由前面的分析可獲得以下的最終狀態。

　　說明：p2線程調用lock.lock後，會禁止p2線程的繼續運行，由於執行了LockSupport.park操做。

　　3. c1線程調用lock.lock，由前面的分析獲得以下的最終狀態。

　　說明：最終c1線程會在sync queue隊列的尾部，而且其結點的前驅結點（包含p2的結點）的waitStatus變爲了SIGNAL。

　　4. c2線程調用lock.lock，由前面的分析獲得以下的最終狀態。

　　說明：最終c1線程會在sync queue隊列的尾部，而且其結點的前驅結點（包含c1的結點）的waitStatus變爲了SIGNAL。

 　　5. p1線程執行emptyCondition.signal，其函數調用順序以下，只給出了主要的函數調用。

　　說明：AQS.CO表示AbstractQueuedSynchronizer.ConditionObject類。此時調用signal方法不會產生任何其餘效果。

　　6. p1線程執行lock.unlock，根據前面的分析可知，最終的狀態以下。

　　說明：此時，p2線程所在的結點爲頭結點，而且其餘兩個線程（c一、c2）依舊被禁止，因此，此時p2線程繼續運行，執行用戶邏輯。

　　7. p2線程執行fullCondition.await，其函數調用順序以下，只給出了主要的函數調用。

　　說明：最終到達的狀態是新生成了一個結點，包含了p2線程，此結點在condition queue中；而且sync queue中p2線程被禁止了，由於在執行了LockSupport.park操做。從函數一些調用可知，在await操做中線程會釋放鎖資源，供其餘線程獲取。同時，head結點後繼結點的包含的線程的許可被釋放了，故其能夠繼續運行。因爲此時，只有c1線程能夠運行，故運行c1。

　　8. 繼續運行c1線程，c1線程因爲以前被park了，因此此時恢復，繼續以前的步驟，即仍是執行前面提到的acquireQueued函數，以後，c1判斷本身的前驅結點爲head，而且能夠獲取鎖資源，最終到達的狀態以下。

　　說明：其中，head設置爲包含c1線程的結點，c1繼續運行。

　　9. c1線程執行fullCondtion.signal，其函數調用順序以下，只給出了主要的函數調用。

　　

　　說明：signal函數達到的最終結果是將包含p2線程的結點從condition queue中轉移到sync queue中，以後condition queue爲null，以前的尾結點的狀態變爲SIGNAL。

　　10. c1線程執行lock.unlock操做，根據以前的分析，經歷的狀態變化以下。

　　

　　說明：最終c2線程會獲取鎖資源，繼續運行用戶邏輯。

　　11. c2線程執行emptyCondition.await，由前面的第七步分析，可知最終的狀態以下。

　　說明：await操做將會生成一個結點放入condition queue中與以前的一個condition queue是不相同的，而且unpark頭結點後面的結點，即包含線程p2的結點。

　　12. p2線程被unpark，故能夠繼續運行，通過CPU調度後，p2繼續運行，以後p2線程在AQS:await函數中被park，繼續AQS.CO:await函數的運行，其函數調用順序以下，只給出了主要的函數調用。

　　13. p2繼續運行，執行emptyCondition.signal，根據第九步分析可知，最終到達的狀態以下。

　　說明：最終，將condition queue中的結點轉移到sync queue中，並添加至尾部，condition queue會爲空，而且將head的狀態設置爲SIGNAL。

　　14. p2線程執行lock.unlock操做，根據前面的分析可知，最後的到達的狀態以下。

　　

　　說明：unlock操做會釋放c2線程的許可，而且將頭結點設置爲c2線程所在的結點。

　　15. c2線程繼續運行，執行fullCondition. signal，因爲此時fullCondition的condition queue已經不存在任何結點了，故其不會產生做用。

　　16. c2執行lock.unlock，因爲c2是sync隊列中最後一個結點，故其不會再調用unparkSuccessor了，直接返回true。即整個流程就完成了。

5、總結

　　對於AbstractQueuedSynchronizer的分析，最核心的就是sync queue的分析。

　　① 每個結點都是由前一個結點喚醒

　　② 當結點發現前驅結點是head而且嘗試獲取成功，則會輪到該線程運行。

　　③ condition queue中的結點向sync queue中轉移是經過signal操做完成的。

　　④ 當結點的狀態爲SIGNAL時，表示後面的結點須要運行。

　　固然，這次分析沒有涉及到中斷操做，若是涉及到中斷操做，又會複雜得多，之後遇到這種狀況，咱們再進行詳細分析，AbstractQueuedSynchronizer類的設計使人歎爲觀止，之後有機會還會進行分析。也謝謝各位園友的觀看~

 

最後給出兩篇參考連接

http://ifeve.com/introduce-abstractqueuedsynchronizer/

http://blog.csdn.net/chen77716/article/details/6641477