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>
 *
复制代码